fn report_timestamp_bindings(&mut self) {
        // Do nothing if dataflow workers can't send feedback or if not enough time has elapsed since
        // the last time we reported timestamp bindings.
        if self.last_bindings_feedback.elapsed().as_millis() < TS_BINDING_FEEDBACK_INTERVAL_MS {
            return;
        }

        let mut changes = Vec::new();
        let mut bindings = Vec::new();
        let mut new_frontier = Antichain::new();

        // Need to go through all sources that are generating timestamp bindings, and extract their upper frontiers.
        // If that frontier is different than the durability frontier we've previously reported then we also need to
        // get the new bindings we've produced and send them to the coordinator.
        for (id, history) in self.render_state.ts_histories.iter() {
            if !history.requires_persistence() {
                continue;
            }

            // Read the upper frontier and compare to what we've reported.
            history.read_upper(&mut new_frontier);
            let prev_frontier = self
                .reported_bindings_frontiers
                .get_mut(&id)
                .expect("Frontier missing!");
            assert!(<_ as PartialOrder>::less_equal(
                prev_frontier,
                &new_frontier
            ));
            if prev_frontier != &new_frontier {
                let mut change_batch = ChangeBatch::new();
                for time in prev_frontier.elements().iter() {
                    change_batch.update(time.clone(), -1);
                }
                for time in new_frontier.elements().iter() {
                    change_batch.update(time.clone(), 1);
                }
                change_batch.compact();
                if !change_batch.is_empty() {
                    changes.push((*id, change_batch));
                }
                // Add all timestamp bindings we know about between the old and new frontier.
                bindings.extend(
                    history
                        .get_bindings_in_range(prev_frontier.borrow(), new_frontier.borrow())
                        .into_iter()
                        .map(|(pid, ts, offset)| (*id, pid, ts, offset)),
                );
                prev_frontier.clone_from(&new_frontier);
            }
        }

        if !changes.is_empty() || !bindings.is_empty() {
            self.send_response(Response::TimestampBindings(TimestampBindingFeedback {
                changes,
                bindings,
            }));
        }
        self.last_bindings_feedback = Instant::now();
    }
/// Send information about new timestamp bindings created by dataflow workers back to
/// the coordinator.
Send information about new timestamp bindings created by dataflow workers back to
the coordinator.
Send information about new timestamp bindings created by dataflow workers back to the coordinator .
fn report_timestamp_bindings(&mut self) {
        // Do nothing if dataflow workers can't send feedback or if not enough time has elapsed since
        // the last time we reported timestamp bindings.
        if self.last_bindings_feedback.elapsed().as_millis() < TS_BINDING_FEEDBACK_INTERVAL_MS {
            return;
        }
        let mut changes = Vec::new();
        let mut bindings = Vec::new();
        let mut new_frontier = Antichain::new();
        // Need to go through all sources that are generating timestamp bindings, and extract their upper frontiers.
        // If that frontier is different than the durability frontier we've previously reported then we also need to
        // get the new bindings we've produced and send them to the coordinator.
        for (id, history) in self.render_state.ts_histories.iter() {
            if !history.requires_persistence() {
                continue;
            }
            // Read the upper frontier and compare to what we've reported.
            history.read_upper(&mut new_frontier);
            let prev_frontier = self
                .reported_bindings_frontiers
                .get_mut(&id)
                .expect("Frontier missing!");
            assert!(<_ as PartialOrder>::less_equal(
                prev_frontier,
                &new_frontier
            ));
            if prev_frontier != &new_frontier {
                let mut change_batch = ChangeBatch::new();
                for time in prev_frontier.elements().iter() {
                    change_batch.update(time.clone(), -1);
                }
                for time in new_frontier.elements().iter() {
                    change_batch.update(time.clone(), 1);
                }
                change_batch.compact();
                if !change_batch.is_empty() {
                    changes.push((*id, change_batch));
                }
                // Add all timestamp bindings we know about between the old and new frontier.
                bindings.extend(
                    history
                        .get_bindings_in_range(prev_frontier.borrow(), new_frontier.borrow())
                        .into_iter()
                        .map(|(pid, ts, offset)| (*id, pid, ts, offset)),
                );
                prev_frontier.clone_from(&new_frontier);
            }
        }
        if !changes.is_empty() || !bindings.is_empty() {
            self.send_response(Response::TimestampBindings(TimestampBindingFeedback {
                changes,
                bindings,
            }));
        }
        self.last_bindings_feedback = Instant::now();
    }
fn report_timestamp_bindings ( & mut self ) { // Do nothing if dataflow workers can't send feedback or if not enough time has elapsed since // the last time we reported timestamp bindings. if self . last_bindings_feedback . elapsed ( ) . as_millis ( ) < TS_BINDING_FEEDBACK_INTERVAL_MS { return ; } let mut changes = Vec :: new ( ) ; let mut bindings = Vec :: new ( ) ; let mut new_frontier = Antichain :: new ( ) ; // Need to go through all sources that are generating timestamp bindings, and extract their upper frontiers. // If that frontier is different than the durability frontier we've previously reported then we also need to // get the new bindings we've produced and send them to the coordinator. for ( id , history ) in self . render_state . ts_histories . iter ( ) { if ! history . requires_persistence ( ) { continue ; } // Read the upper frontier and compare to what we've reported. history . read_upper ( & mut new_frontier ) ; let prev_frontier = self . reported_bindings_frontiers . get_mut ( & id ) . expect ( " " ) ; assert ! ( as PartialOrder less_equal ( prev_frontier new_frontier ) ) ; if prev_frontier != & new_frontier { let mut change_batch = ChangeBatch :: new ( ) ; for time in prev_frontier . elements ( ) . iter ( ) { change_batch . update ( time . clone ( ) , - 1 ) ; } for time in new_frontier . elements ( ) . iter ( ) { change_batch . update ( time . clone ( ) , 1 ) ; } change_batch . compact ( ) ; if ! change_batch . is_empty ( ) { changes . push ( ( * id , change_batch ) ) ; } // Add all timestamp bindings we know about between the old and new frontier. bindings . extend ( history . get_bindings_in_range ( prev_frontier . borrow ( ) , new_frontier . borrow ( ) ) . into_iter ( ) . map ( | ( pid , ts , offset ) | ( * id , pid , ts , offset ) ) , ) ; prev_frontier . clone_from ( & new_frontier ) ; } } if ! changes . is_empty ( ) || ! bindings . is_empty ( ) { self . send_response ( Response :: TimestampBindings ( TimestampBindingFeedback { changes , bindings , } ) ) ; } self . last_bindings_feedback = Instant :: now ( ) ; }
Send information about new timestamp bindings created by dataflow workers back to
the coordinator.
Send information about new timestamp bindings created by dataflow workers back to the coordinator .
&mut self
&mut self