Release preparation (#722)

* Slow but correct impls

* WIP

* Wildly overcomplicated

* Remove commented code

* Removed one-off trait

* Remove unread counters

* Convert silent errors to panics

* Simplify logic

* More commented code removed

* Extract unconditional behavior

* Idiomatic Rust

* Less time cloning

* Idiomatic capability use

* Ref keyword to borrow

* Further tightening

* Tidy explanatory text, and reflect frontier

* Tighten comments, remove mutable borrow

* Prefer insert_ref to insert

* Use containers for interesting (keys, time) (#710)

* Use containers for interesting (keys, time)

* Remove use of owned keys in reduce.rs

* Remove all uses of KeyOwn

* Remove KeyOwn

* Improve documentation

* Walk back overly prescriptive dogs^3 constraints

* Remove TimelyStack and all dependent types (#715)

Remove TimelyStack, TStack layout, ColumnationChunker, ColInternalMerger,
and all Col* type aliases (ColValSpine, ColKeySpine, ColValBuilder, etc.)
from the codebase. The columnation crate dependency is retained.

Co-authored-by: Claude Opus 4.6 (1M context) <noreply@anthropic.com>

* Track timely `master` changes, around `Child` scope changes (#714)

* Track timely's Child changes

* Tidy bounds

* Absorb next wave of changes

* Remove unused, and Arranged impls

* Remove timestamp generic from Arranged

* Standardize T for timestamps, Tr for traces

* Correct local timely reference

* Convert G generics to T

* Remove as Timestamp prompts

* Remove T generics bound by Tr

* Remove Scope::clone calls

* Further tightening of traits

* Correct docs

* Relifetimed scopes (#718)

* Bring core DD into line

* Clean up examples and tests

* Interactive update (#719)

* Example DDIR with SCC

* Parse from concrete syntax

* Further examples and improvements

* Further improvements to track columnar

* Move DDIR to its own crate

Extract the DD IR interpreter into a `ddir/` crate, separating shared
infrastructure (parse, IR, lower) from backend-specific rendering.

- Split parse into `parse/applicative.rs` (.ddir) and `parse/pipe.rs` (.ddp)
- Extract `ir.rs` (Node, Program, RowLike) from `lower.rs`
- Move programs and examples from differential-dataflow/examples/ to ddir/
- Remove DDIR_DEBUG, DDIR_REACHABILITY, DDIR_PRINT env vars (use INSPECT)
- Fix off-by-one in columnar harness timestamping
- Unify inspect output format between vec and col backends
- Update programs to output compact counts via arrange+inspect

Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>

* Optimized IR

* Relocate to interactive/

* Further clean-up

---------

Co-authored-by: Claude Opus 4.6 (1M context) <noreply@anthropic.com>

* DDIR README

* Continue to track timely master (#720)

* Target timely 0.29 (#721)

---------

Co-authored-by: Moritz Hoffmann <antiguru@gmail.com>
Co-authored-by: Claude Opus 4.6 (1M context) <noreply@anthropic.com>

Author: 3 people

Cargo.toml

@@ -23,9 +23,9 @@ rust-version = "1.86"
 
 [workspace.dependencies]
 differential-dataflow = { path = "differential-dataflow", default-features = false, version = "0.22.0" }
-timely = { version = "0.28", default-features = false }
+timely = { version = "0.29", default-features = false }
 columnar = { version = "0.12", default-features = false }
-#timely = { git = "https://github.com/TimelyDataflow/timely-dataflow" }
+#timely = { git = "https://github.com/TimelyDataflow/timely-dataflow", default-features = false }
 #timely = { path = "../timely-dataflow/timely/", default-features = false }
 
 [workspace.lints.clippy]

README.md

@@ -22,7 +22,7 @@ Alternately, here is a fragment that computes the set of nodes reachable from a
 ```rust
 let reachable =
 roots.iterate(|scope, reach|
-    edges.enter(&scope)
+    edges.enter(scope)
          .semijoin(reach)
          .map(|(src, dst)| dst)
          .concat(reach)
@@ -337,7 +337,7 @@ edges.iterate(|scope, inner| {
                       .map(|(node,_)| node);
 
     // keep edges between active vertices
-    edges.enter(&scope)
+    edges.enter(scope)
          .semijoin(active)
          .map(|(src,dst)| (dst,src))
          .semijoin(active)

advent_of_code_2017/Cargo.toml

@@ -11,4 +11,4 @@ workspace = true
 
 [dependencies]
 differential-dataflow = { workspace = true }
-timely = { git = "https://github.com/frankmcsherry/timely-dataflow" }
+timely = { workspace = true }

advent_of_code_2017/src/bin/day_06.rs

@@ -24,7 +24,7 @@ fn main() {
 
             let stable = banks.iterate(|scope, iter|
                 iter.map_in_place(|banks| recycle(banks))
-                    .concat(banks.enter(&scope))
+                    .concat(banks.enter(scope))
                     .distinct()
             );
 
@@ -43,7 +43,7 @@ fn main() {
             loop_point
                 .iterate(|scope, iter|
                     iter.map_in_place(|banks| recycle(banks))
-                        .concat(loop_point.enter(&scope))
+                        .concat(loop_point.enter(scope))
                         .distinct()
                 )
                 .map(|_| ((),()))

advent_of_code_2017/src/bin/day_07.rs

@@ -1104,10 +1104,10 @@ tvhftq (35)";
 
             let total_weights: VecCollection<_,String> = weights
                 .iterate(|scope, inner| {
-                    parents.enter(&scope)
+                    parents.enter(scope)
                            .semijoin(inner)
                            .map(|(_, parent)| parent)
-                           .concat(weights.enter(&scope))
+                           .concat(weights.enter(scope))
                 });
 
             parents

advent_of_code_2017/src/bin/day_08.rs

@@ -1107,7 +1107,7 @@ wui inc -120 if i > -2038";
                 .map(|_| ((0, String::new()), 0))
                 .iterate(|scope, valid| {
 
-                    let edits = edits.enter(&scope);
+                    let edits = edits.enter(scope);
 
                     valid
                         .prefix_sum_at(edits.map(|(key,_)| key), 0, |_k,x,y| *x + *y)

advent_of_code_2017/src/bin/day_09.rs

@@ -120,7 +120,7 @@ where
                     if input.len() > 1 { result = combine(result, &(input[1].0).1); }
                     output.push((result, 1));
                 })
-                .concat(unit_ranges.enter(&scope))
+                .concat(unit_ranges.enter(scope))
         )
 }
 
@@ -154,10 +154,10 @@ where
         .iterate(|scope, state| {
             aggregates
                 .filter(|&((_, log),_)| log < 64)    // the log = 64 interval doesn't help us here (overflows).
-                .enter(&scope)
+                .enter(scope)
                 .map(|((pos, log), data)| (pos, (log, data)))
                 .join_map(state, move |&pos, &(log, ref data), state| (pos + (1 << log), combine(state, data)))
-                .concat(init_state.enter(&scope))
+                .concat(init_state.enter(scope))
                 .distinct()
         })
         .consolidate()

advent_of_code_2017/src/bin/day_12.rs

@@ -2035,8 +2035,8 @@ fn main() {
             let labels =
             nodes
                 .iterate(|scope, label| {
-                    let edges = edges.enter(&scope);
-                    let nodes = nodes.enter(&scope);
+                    let edges = edges.enter(scope);
+                    let nodes = nodes.enter(scope);
                     label
                         .join_map(edges, |_src, &lbl, &tgt| (tgt, lbl))
                         .concat(nodes)

diagnostics/Cargo.toml

@@ -1,6 +1,7 @@
 [package]
 name = "diagnostics"
 version = "0.1.0"
+publish = false
 edition.workspace = true
 rust-version.workspace = true
 

diagnostics/src/logging.rs

@@ -34,7 +34,6 @@ use differential_dataflow::operators::arrange::TraceAgent;
 use differential_dataflow::trace::implementations::{KeySpine, ValSpine};
 use differential_dataflow::{AsCollection, VecCollection};
 
-use timely::communication::Allocate;
 use timely::container::CapacityContainerBuilder;
 use timely::dataflow::channels::pact::Pipeline;
 use timely::dataflow::operators::capture::{Event, EventLink, Replay, Capture};
@@ -244,12 +243,11 @@ fn quantize(time: Duration, interval: Duration) -> Duration {
 }
 
 /// Quantize timestamps in a collection's inner stream.
-fn quantize_collection<S, D>(
-    collection: VecCollection<S, D, i64>,
+fn quantize_collection<'scope, D>(
+    collection: VecCollection<'scope, Duration, D, i64>,
     interval: Duration,
-) -> VecCollection<S, D, i64>
+) -> VecCollection<'scope, Duration, D, i64>
 where
-    S: Scope<Timestamp = Duration>,
     D: differential_dataflow::Data,
 {
     collection
@@ -276,7 +274,7 @@ where
 ///
 /// Returns a [`LoggingState`] with trace handles and a [`SinkHandle`] for
 /// the WebSocket thread.
-pub fn register<A: Allocate>(worker: &mut Worker<A>, log_logging: bool) -> LoggingState {
+pub fn register(worker: &mut Worker, log_logging: bool) -> LoggingState {
     let start = Instant::now();
 
     // Event links for logging capture (worker-internal, Rc-based).
@@ -377,8 +375,8 @@ pub fn register<A: Allocate>(worker: &mut Worker<A>, log_logging: bool) -> Loggi
     }
 }
 
-fn install_loggers<A: Allocate>(
-    worker: &mut Worker<A>,
+fn install_loggers(
+    worker: &mut Worker,
     t_link: Rc<EventLink<Duration, Vec<(Duration, TimelyEvent)>>>,
     d_link: Rc<EventLink<Duration, Vec<(Duration, DifferentialEvent)>>>,
 ) {
@@ -402,11 +400,11 @@ fn install_loggers<A: Allocate>(
 // ============================================================================
 
 /// Internal: collections before arrangement, used for the cross-join.
-struct TimelyCollections<S: Scope> {
-    operators: VecCollection<S, (usize, String, Vec<usize>), i64>,
-    channels: VecCollection<S, (usize, Vec<usize>, (usize, usize), (usize, usize)), i64>,
-    elapsed: VecCollection<S, usize, i64>,
-    messages: VecCollection<S, usize, i64>,
+struct TimelyCollections<'scope> {
+    operators: VecCollection<'scope, Duration, (usize, String, Vec<usize>), i64>,
+    channels: VecCollection<'scope, Duration, (usize, Vec<usize>, (usize, usize), (usize, usize)), i64>,
+    elapsed: VecCollection<'scope, Duration, usize, i64>,
+    messages: VecCollection<'scope, Duration, usize, i64>,
 }
 
 #[derive(Default)]
@@ -416,16 +414,16 @@ struct TimelyDemuxState {
 }
 
 /// Build timely logging collections and arrangements.
-fn construct_timely<S: Scope<Timestamp = Duration>>(
-    scope: &mut S,
-    stream: Stream<S, Vec<(Duration, TimelyEvent)>>,
-) -> (TimelyTraces, TimelyCollections<S>) {
+fn construct_timely<'scope>(
+    scope: Scope<'scope, Duration>,
+    stream: Stream<'scope, Duration, Vec<(Duration, TimelyEvent)>>,
+) -> (TimelyTraces, TimelyCollections<'scope>) {
     type OpUpdate = ((usize, String, Vec<usize>), Duration, i64);
     type ChUpdate = ((usize, Vec<usize>, (usize, usize), (usize, usize)), Duration, i64);
     type ElUpdate = (usize, Duration, i64);
     type MsgUpdate = (usize, Duration, i64);
 
-    let mut demux = OperatorBuilder::new("Timely Demux".to_string(), scope.clone());
+    let mut demux = OperatorBuilder::new("Timely Demux".to_string(), scope);
     let mut input = demux.new_input(stream, Pipeline);
 
     let (op_out, operates) = demux.new_output::<Vec<OpUpdate>>();
@@ -536,24 +534,24 @@ fn construct_timely<S: Scope<Timestamp = Duration>>(
 // ============================================================================
 
 /// Internal: collections before arrangement, used for the cross-join.
-struct DifferentialCollections<S: Scope> {
-    arrangement_batches: VecCollection<S, usize, i64>,
-    arrangement_records: VecCollection<S, usize, i64>,
-    sharing: VecCollection<S, usize, i64>,
-    batcher_records: VecCollection<S, usize, i64>,
-    batcher_size: VecCollection<S, usize, i64>,
-    batcher_capacity: VecCollection<S, usize, i64>,
-    batcher_allocations: VecCollection<S, usize, i64>,
+struct DifferentialCollections<'scope> {
+    arrangement_batches: VecCollection<'scope, Duration, usize, i64>,
+    arrangement_records: VecCollection<'scope, Duration, usize, i64>,
+    sharing: VecCollection<'scope, Duration, usize, i64>,
+    batcher_records: VecCollection<'scope, Duration, usize, i64>,
+    batcher_size: VecCollection<'scope, Duration, usize, i64>,
+    batcher_capacity: VecCollection<'scope, Duration, usize, i64>,
+    batcher_allocations: VecCollection<'scope, Duration, usize, i64>,
 }
 
 /// Build differential logging collections and arrangements.
-fn construct_differential<S: Scope<Timestamp = Duration>>(
-    scope: &mut S,
-    stream: Stream<S, Vec<(Duration, DifferentialEvent)>>,
-) -> (DifferentialTraces, DifferentialCollections<S>) {
+fn construct_differential<'scope>(
+    scope: Scope<'scope, Duration>,
+    stream: Stream<'scope, Duration, Vec<(Duration, DifferentialEvent)>>,
+) -> (DifferentialTraces, DifferentialCollections<'scope>) {
     type Update = (usize, Duration, i64);
 
-    let mut demux = OperatorBuilder::new("Differential Demux".to_string(), scope.clone());
+    let mut demux = OperatorBuilder::new("Differential Demux".to_string(), scope);
     let mut input = demux.new_input(stream, Pipeline);
 
     let (bat_out, batches) = demux.new_output::<Vec<Update>>();