fix: frontier freezes when updates in ValDistributor consolidate away

[ParkMyCar]

Hey Frank!

I ran into an issue with the columnar ValDistributor that caused my dataflow to hang, i.e. the frontier of an operator stopped advancing.

As far as I (and Claude) understand, timely tracks the number of records emitted by an operator and all records consumed by the next downstream operator. timely considers produced - consumed > 0 as meaning "records still in flight" and thus it cannot advance the frontier.

Issue

The RecordedUpdates container used by ValDistributor consolidates as records get pushed into the trie. The problem is when all of the per-worker outputs consolidate to nothing. ValDistributor emits no messages and thus timely forever sees outstanding records on that edge.

Fix

This PR adds a check for this specific case, and emits a single update whose record count is the total number of input records. It also includes a regression test that exhibits the problem on today's master.

[frankmcsherry]

I'll check; you are right that there is an invariant that one needs to ship "the number of expected records" not "what they end up being once you compact them", or at least an int saying as much. Brb once I point some robots at it.

[frankmcsherry]

Yup, looks good and makes sense. The only nuance I would add is that the columnar/ container here is .. mostly kicking the tires on some abstractions, and may not be production ready yet (stronger: is not production ready). If you are just screwing around then no worries, but if it's more serious than that it may be worth chatting briefly (MZ has a different columnar container it uses, and the repo's direction is more in the interactive/ direction with an interpreted columnar layer; but it all could/should work, just may not yet).

[ParkMyCar]

Thanks for the quick review, and heads up about the status of the columnar/ module!

Mostly just experimenting at the moment, but if it goes further I would love to chat.

[frankmcsherry]

Sounds great; and thank you for the bug report and fix!