[dbsp] Skip a recursive sub-circuit at a step where its inputs did not change #7744

Summary

A recursive sub-circuit runs a full fixpoint at every step of its parent circuit, also when none of its imported inputs changed. In transaction mode one commit takes many root steps, so most of these runs do no useful work. In one customer pipeline, these idle runs are about 83% of the commit time (15 s of 18 s for a 4 to 15 row transaction).

Proposal: when no worker has a change on any import of the sub-circuit, and the sub-circuit has no pending work, do not run the nested transaction. Write the zero result to the export streams, and report the flush as complete.

Current behavior

Result: an idle run costs (inner receiver chain + 1) inner steps. Each inner step evaluates all operators of the sub-circuit and ends with a consensus across all workers.

Evidence

Source: a customer support bundle, v0.358.0, 8 workers, 1 host. The program has 113 views and 6 recursive groups. Input comes through HTTP ingress transactions of 4 to 15 rows. The window holds 8 commits and 761 root steps (95 per commit). Times are from the slowest worker.

• Each group runs at every root step: runs = 761 = root steps.
• Group D got 0 input rows in the window. It still cost 2.2 s per commit.
• The measured iterations per run match (inner chain + 1) for all 6 groups. A and B are higher because some runs had data and needed more than one inner transaction.
• Model: commit = (root chain + 1) x (root step cost + sum over groups of (inner chain + 1) x inner step cost) = 95 x (0.033 s + 0.158 s) = 18.1 s.

Expected effect (estimate, not measured)

We expect each import to carry data at a small number of root steps per commit (k), because the accumulator before it emits once per commit. If each group runs only at those k steps:

The bundle cannot show k directly. #7743 adds profile counters for it (subcircuit_runs_count, subcircuit_runs_with_input_count) and a DBSP test, run_stats::recursion_runs_at_every_root_step. In that test, with 4 workers and a chain of 8 joins, a commit that does not touch the recursion's input runs the recursion at all of its 7 to 9 root steps, with no input on any worker. A commit that does touch it gives the recursion input in 1 of 9 runs. So k = 1 in the test.

Proposal

In ChildNode::eval (or in IterativeExecutor):

1. After the import nodes run, each worker checks if all of its imports are empty in this step. For Delta0 this means an empty batch or no value.
2. All workers agree on one boolean: if any worker has a change, all workers run. The decision must be the same on all workers, because the inner exchanges and the inner commit consensus need every worker. One consensus round replaces (inner chain + 1) inner steps.
3. If no worker has a change, the sub-circuit is at its fixpoint from the last run, and no inner operator has pending work, then:
   • do not call executor.transaction();
   • write the zero value to each export stream. Today FeedbackOutputNode::clock_end writes it (circuit_builder.rs#L6881-L6887). If nothing writes it, the parent reads a stale value or no value;
   • if the parent flushed the ChildNode, report the flush as complete.

Things to check

• Clock. clock_end advances the nested clock (circuit_builder.rs#L3863-L3870). Inner traces use it for timestamps, and the sub-circuit checkpoint stores it (circuit_builder.rs#L7402). A skipped run does not advance it. Confirm that inner traces and checkpoint restore accept this.
• Imports other than Delta0, for example traces of the parent, need their own test for "changed".
• Bootstrap and replay of recursive views must not skip. This includes concurrent bootstrap and swap_state, where the nested clock is realigned.
• Inner operators that report fixedpoint() == false at the end of a run, or that have pending output or are async and not ready, must force a run.
• Multihost: the agreement in step 2 must include all hosts.

Acceptance criteria

• A transaction that does not change the inputs of a recursive view does not start a nested transaction for it. The steps_count of that scope in the circuit profile does not grow, except at the steps where its imports have data.
• The outputs do not change. The existing recursive tests pass. Add a test that commits transactions which do not touch the recursion's inputs, and check the scope's step count and the outputs.
• Works with more than one worker and with more than one host.

Related

• A shuffle hop costs one step at commit, for the same reason as step 4 above. This sets the number of root steps (95 here), and the number of inner steps per run. ShardedAccumulator::wait_for_arrival (sharded_accumulator.rs#L663-L664) has no production caller yet. That is a separate change, tracked in [dbsp] Let a sharded accumulator receiver finish its flush in the step in which its peers flush #7745. This issue is only about not running a recursion that has no new input.
• [dbsp] Count the runs of a nested circuit that get no input #7743 adds the counters and the test that measure the idle runs.