storage: make decode_and_mfp yield more often

We've seen evidence in production of this operator often going more than
5s without yielding, especially during rehydration and large unindexed
SELECTs. This affects interactivity as well as the time between when a
timely-driven Future is woken and when it is scheduled again.

Attempt to fix this with two tweaks to our yield heuristics. First,
decrease the fuel between yields from 1_000_000 to 100_000. Second, also
count each mfp evaluation against our fuel, in case it contains a
restrictive filter.

The new 100_000 constant comes from eyeballing a replica in prod that
was non-interactive. At the time, it was decoding ~300k rows per second,
so 100k doesn't seem terribly low in comparison.

Author: danhhz

src/compute/src/render/mod.rs

@@ -234,8 +234,9 @@ pub fn build_compute_dataflow<A: Allocate>(
                         dataflow.until.clone(),
                         mfp.as_mut(),
                         Some(flow_control),
-                        // Copy the logic in DeltaJoin/Get/Join to start.
-                        |_timer, count| count > 1_000_000,
+                        // Tuned down from 1_000_000 (the original logic copied from
+                        // DeltaJoin/Get/Join) to increase interactivity.
+                        |_timer, count| count > 100_000,
                     );
 
                     // If `mfp` is non-identity, we need to apply what remains.

src/compute/src/sink/persist_sink.rs

@@ -102,8 +102,9 @@ where
         Antichain::new(), // we want all updates
         None,             // no MFP
         None,             // no flow control
-        // Copy the logic in DeltaJoin/Get/Join to start.
-        |_timer, count| count > 1_000_000,
+        // Tuned down from 1_000_000 (the original logic copied from DeltaJoin/Get/Join) to increase
+        // interactivity.
+        |_timer, count| count > 100_000,
     );
     use differential_dataflow::AsCollection;
     let persist_collection = ok_stream

src/storage-operators/src/persist_source.rs

@@ -365,6 +365,13 @@ impl PendingWork {
             match (key, val) {
                 (Ok(SourceData(Ok(row))), Ok(())) => {
                     if let Some(mfp) = map_filter_project {
+                        // We originally accounted work as the number of outputs, to give downstream
+                        // operators a chance to reduce down anything we've emitted. This mfp call
+                        // might have a restrictive filter, which would have been counted as no
+                        // work. However, in practice, we've been decode_and_mfp be a source of
+                        // interactivity loss during rehydration, so we now also count each mfp
+                        // evaluation against our fuel.
+                        *work += 1;
                         let arena = mz_repr::RowArena::new();
                         let mut datums_local = datum_vec.borrow_with(&row);
                         for result in mfp.evaluate(

src/storage/src/render/sinks.rs

@@ -56,8 +56,9 @@ pub(crate) fn render_sink<'g, G: Scope<Timestamp = ()>>(
         timely::progress::Antichain::new(),
         None,
         None,
-        // Copy the logic in DeltaJoin/Get/Join to start.
-        |_timer, count| count > 1_000_000,
+        // Tuned down from 1_000_000 (the original logic copied from DeltaJoin/Get/Join) to increase
+        // interactivity.
+        |_timer, count| count > 100_000,
     );
     tokens.push(source_token);
 

src/storage/src/render/sources.rs

@@ -359,8 +359,9 @@ where
                                     Antichain::new(),
                                     None,
                                     None,
-                                    // Copy the logic in DeltaJoin/Get/Join to start.
-                                    |_timer, count| count > 1_000_000,
+                                    // Tuned down from 1_000_000 (the original logic copied from
+                                    // DeltaJoin/Get/Join) to increase interactivity.
+                                    |_timer, count| count > 100_000,
                                 );
                             let (tx_source_ok, tx_source_err) = (
                                 tx_source_ok_stream.as_collection(),