deps: build against differential-dataflow master-next and timely master

[antiguru]

Draft. Front-loads the adjustment to the next differential-dataflow, and records what it costs us, so the work is written down rather than discovered under time pressure when it lands. Not merge-ready, for reasons listed at the bottom.

Builds against TimelyDataflow/differential-dataflow master-next at 229508dd, which pins timely by git, so timely master comes along. That is 70 differential commits of delta, of which roughly 40 are Corgi/DDIR/formal work with no effect here.

What moved

Batch and BatchReader are gone. A trace's batch is bare 'static + Clone, and the description that used to hang off it lives on the new Span<T, B>, whose batch is absent exactly when the interval carried no updates. TraceReader loses cursor/cursor_through for spans_through/batches_through plus trace::cursor::cursor_list, and map_batches becomes map_spans. SpineBatch shrinks to Time, Merger, len, which takes a good deal of ArcBatch with it.

Builder becomes Default + push + done(self) -> Option<Output>, sealing moves out to Sealer, and Batcher takes its input container as a trait parameter, owns its chunker, and returns (Option<Output>, AntichainRef). So MergeBatcher names chunker, merger and sealer together, and mz_arrange::<Chu, Ba, Bu, Tr> collapses to mz_arrange::<Ba, Tr> — those three parameters were always one unit, and differential now says so.

Consolidation wants the chunks rather than a trace batch assembled from them and immediately taken apart, so ConsolidatingBatcher is a Batcher whose output is the chain, and ColumnMergeBatcher grows extract_chain with its Batcher::extract being that plus a seal.

What it deletes

Net -734 lines. Three forks stop earning their keep:

• mz_join_core, which existed to yield within keys. The differential join driver owns a fuel budget and yields between output containers, so join_core covers it, and LinearJoinSpec/YieldSpec go with it.
• build_halfjoin1 in the delta join, since upstream folded half_join2 into half_join. half_join_internal_unsafe now takes the tie-break as a flag where we passed a comparison closure — which was already a bool one frame up, as source_precedes_lookup.
• UpsertFeedbackBatcher, a wrapper type whose only job was injecting the storage pager into a constructor the trait did not expose. Now a closure at the arrange_core call site.

That leaves enable_mz_join_core, linear_join_yielding and enable_half_join2 with no readers. Removing the definitions is a follow-up.

Timely's multi-capability stamps

A message's capabilities are a Stamp, a multiset. InputCapability::time, retain and Deref panic unless it is a singleton, and nothing about that is visible at compile time. A clippy.toml disallowed-methods entry for the first two is the only mechanical guard, and it found sites that reading the code had missed, including builder_async's InputConnection::accept — the funnel every async operator's capability comes through.

Non-singleton stamps have two sources: a partially ordered timestamp, where the antichain of capabilities an operator retires can hold more than one element; and a scope-boundary map, which Stamp::map_pointwise makes count-preserving on purpose, so collapsing inner times produce duplicates. A T: TotalOrder bound therefore does not imply a singleton on its own.

Fixed here: ArrangementFlatMap and its ok-only twin hold a CapabilitySet (so the Session alias is typed over one), TopKIntraTimeThinning buckets by the record's own time rather than the message's, and LimitProgress keeps an antichain instead of one hand-picked capability. Every remaining site carries an allow saying why its stamp is a singleton.

Tests

No new test files. render/top_k.rs's intra-time thinning changed shape, so topk.slt covers it; the batcher and builder rewrites are covered by the existing mz-timely-util, mz-row-spine and mz-compute unit tests, whose helpers were rewritten for the new Builder/Batcher APIs.

Why this is a draft

1. bin/lint fails cargo deny sources. deny.toml allows git dependencies only from the MaterializeInc organization, with a rationale about shared maintainership and historical buildability. That policy is right and should not be relaxed for this. The merge path is a published release or MaterializeInc forks — and in any case a branch = reference is not reproducible and would need to become a rev =.

2. Recursive dataflows panic, on an upstream bug rather than anything here. leave_dynamic reads a single time off a multi-capability message, in both implementations:

   • differential-dataflow/src/dynamic/mod.rs:50
   • differential-dataflow/src/columnar/collection/operators.rs:120

   A dynamic scope is exactly where non-singleton stamps arise, so the operator is unreachable rather than failing in a corner case. distinct_arrangements.slt and chbench.slt panic within seconds; with the patch below applied they run. The fix must minimize into a CapabilitySet rather than preserve counts — leave_dynamic is an ordinary operator acquiring fresh capabilities via delayed, not a scope boundary, so minimizing is legitimate there, and downstream async operators depend on it.

   Patch against 229508dd

   --- a/differential-dataflow/src/dynamic/mod.rs
   +++ b/differential-dataflow/src/dynamic/mod.rs
   @@ -47,11 +47,17 @@
            builder.build(move |_capability| move |_frontier| {
                let mut output = output.activate();
                input.for_each(|cap, data| {
   -                let mut new_time = cap.time().clone();
   -                let mut vec = std::mem::take(&mut new_time.inner).into_inner();
   -                vec.truncate(level - 1);
   -                new_time.inner = PointStamp::new(vec);
   -                let new_cap = cap.delayed(&new_time, 0);
   +                // A message carries a multiset of capabilities, so truncate each of them. The
   +                // input's summary is the same truncation, which is what lets `delayed` move a
   +                // capability down to the shorter time.
   +                let mut new_cap = timely::dataflow::operators::CapabilitySet::new();
   +                for time in cap.stamp().iter() {
   +                    let mut new_time = time.clone();
   +                    let mut vec = std::mem::take(&mut new_time.inner).into_inner();
   +                    vec.truncate(level - 1);
   +                    new_time.inner = PointStamp::new(vec);
   +                    new_cap.insert(cap.delayed(&new_time, 0));
   +                }

   The same change applies to columnar/collection/operators.rs:120.

3. Wrapping builders can no longer pre-size. OrdValBuilder::with_capacity and OrdKeyBuilder::with_capacity are private upstream, so the six spine builders compute their key/val/update counts and throw them away. That is an allocation regression and the main thing to raise with upstream alongside the leave_dynamic fix.

Based on ad7b551153; wants a rebase before it goes anywhere.

🤖 Generated with Claude Code