Skip to content

reduce: name the shared key container to fix HRTB normalization #797

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
frankmcsherry merged 1 commit into from
Jul 14, 2026
Merged

reduce: name the shared key container to fix HRTB normalization #797

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
1 change: 1 addition & 0 deletions differential-dataflow/examples/columnar/main.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -171,6 +171,7 @@ mod reachability {
ValBuilder<Node, (), IterTime, Diff>,
ValSpine<Node, (), IterTime, Diff>,
_,
_,
>("Distinct", |_node, _input, output| { output.push(((), 1)); },
|col, key, upds| {
use columnar::Push;
Expand Down
8 changes: 4 additions & 4 deletions differential-dataflow/src/collection.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -749,7 +749,7 @@ pub mod vec {
use crate::trace::implementations::{ValBuilder, ValSpine};

self.arrange_by_key_named(&format!("Arrange: {}", name))
.reduce_abelian::<_,ValBuilder<_,_,_,_>,ValSpine<K,V2,_,_>,_>(
.reduce_abelian::<_,ValBuilder<_,_,_,_>,ValSpine<K,V2,_,_>,_,_>(
name,
logic,
|vec, key, upds| { vec.clear(); vec.extend(upds.drain(..).map(|(v,t,r)| ((key.clone(), v),t,r))); },
Expand Down Expand Up @@ -809,7 +809,7 @@ pub mod vec {
L: FnMut(&K, &[(&V, R)], &mut Vec<(V,BatchDiff<T2>)>, &mut Vec<(V, BatchDiff<T2>)>)+'static,
{
self.arrange_by_key_named(&format!("Arrange: {}", name))
.reduce_core::<_,Bu,_,_>(
.reduce_core::<_,Bu,_,_,_>(
name,
logic,
|vec, key, upds| { vec.clear(); vec.extend(upds.drain(..).map(|(v,t,r)| ((key.clone(), v),t,r))); },
Expand Down Expand Up @@ -878,7 +878,7 @@ pub mod vec {
use crate::trace::implementations::{KeyBuilder, KeySpine};

self.arrange_by_self_named(&format!("Arrange: {}", name))
.reduce_abelian::<_,KeyBuilder<K,T,R2>,KeySpine<K,T,R2>,_>(
.reduce_abelian::<_,KeyBuilder<K,T,R2>,KeySpine<K,T,R2>,_,_>(
name,
move |k,s,t| t.push(((), thresh(k, &s[0].1))),
|vec, key, upds| { vec.clear(); vec.extend(upds.drain(..).map(|(v,t,r)| ((key.clone(), v),t,r))); },
Expand Down Expand Up @@ -919,7 +919,7 @@ pub mod vec {
pub fn count_core<R2: Ord + Abelian + From<i8> + 'static>(self) -> Collection<'scope, T, (K, R), R2> {
use crate::trace::implementations::{ValBuilder, ValSpine};
self.arrange_by_self_named("Arrange: Count")
.reduce_abelian::<_,ValBuilder<K,R,T,R2>,ValSpine<K,R,T,R2>,_>(
.reduce_abelian::<_,ValBuilder<K,R,T,R2>,ValSpine<K,R,T,R2>,_,_>(
"Count",
|_k,s,t| t.push((s[0].1.clone(), R2::from(1i8))),
|vec, key, upds| { vec.clear(); vec.extend(upds.drain(..).map(|(v,t,r)| ((key.clone(), v),t,r))); },
Expand Down
2 changes: 1 addition & 1 deletion differential-dataflow/src/operators/arrange/agent.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -198,7 +198,7 @@
/// // create a second dataflow
/// worker.dataflow(move |scope| {
/// trace.import(scope)
/// .reduce_abelian::<_,ValBuilder<_,_,_,_>,ValSpine<_,_,_,_>,_>(
/// .reduce_abelian::<_,ValBuilder<_,_,_,_>,ValSpine<_,_,_,_>,_,_>(
/// "Reduce",
/// |_key, src, dst| dst.push((*src[0].0, 1)),
/// |vec, key, upds| { vec.clear(); vec.extend(upds.drain(..).map(|(v,t,r)| ((key.clone(), v),t,r))); },
Expand Down Expand Up @@ -283,7 +283,7 @@
let activator = scope.activator_for(Rc::clone(&info.address));
let queue = self.new_listener(activator);

let activator = scope.activator_for(info.address);

Check warning on line 286 in differential-dataflow/src/operators/arrange/agent.rs

View workflow job for this annotation

GitHub Actions / Cargo clippy 

`activator` shadows a previous, unrelated binding
*shutdown_button_ref = Some(ShutdownButton::new(Rc::clone(&capabilities), activator));

capabilities.borrow_mut().as_mut().unwrap().insert(capability);
Expand Down Expand Up @@ -416,7 +416,7 @@
let activator = scope.activator_for(Rc::clone(&info.address));
let queue = self.new_listener(activator);

let activator = scope.activator_for(info.address);

Check warning on line 419 in differential-dataflow/src/operators/arrange/agent.rs

View workflow job for this annotation

GitHub Actions / Cargo clippy 

`activator` shadows a previous, unrelated binding
*shutdown_button_ref = Some(ShutdownButton::new(Rc::clone(&capabilities), activator));

capabilities.borrow_mut().as_mut().unwrap().insert(capability);
Expand Down
29 changes: 17 additions & 12 deletions differential-dataflow/src/operators/arrange/arrangement.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -205,7 +205,7 @@
while let Some(key) = cursor.get_key(batch) {
while let Some(val) = cursor.get_val(batch) {
for datum in logic(key, val) {
cursor.map_times(batch, |time, diff| {

Check warning on line 208 in differential-dataflow/src/operators/arrange/arrangement.rs

View workflow job for this annotation

GitHub Actions / Cargo clippy 

`time` shadows a previous, unrelated binding
session.give((datum.clone(), <BatchCursor<Tr> as Cursor>::owned_time(time), <BatchCursor<Tr> as Cursor>::owned_diff(diff)));
});
}
Expand Down Expand Up @@ -261,18 +261,21 @@

// Direct reduce implementations.
use crate::difference::Abelian;
use crate::trace::implementations::containers::BatchContainer;
impl<'scope, Tr1: TraceReader<Batch: Navigable>+'static> Arranged<'scope, Tr1> {
/// A direct implementation of `ReduceCore::reduce_abelian`.
pub fn reduce_abelian<L, Bu, Tr2, P>(self, name: &str, mut logic: L, push: P) -> Arranged<'scope, TraceAgent<Tr2>>
pub fn reduce_abelian<L, Bu, Tr2, KC, P>(self, name: &str, mut logic: L, push: P) -> Arranged<'scope, TraceAgent<Tr2>>
where
Tr2: Trace<Batch: Navigable, Time=Tr1::Time>+'static,
BatchCursor<Tr1>: Cursor<Time = Tr1::Time>,
for<'a> BatchCursor<Tr2>: Cursor<Key<'a> = BatchKey<'a, Tr1>, ValOwn: Data, Time = Tr2::Time, Diff: Abelian>,
KC: BatchContainer,
BatchCursor<Tr1>: Cursor<Time = Tr1::Time, KeyContainer = KC>,
for<'a> BatchCursor<Tr1>: Cursor<Key<'a> = KC::ReadItem<'a>>,
for<'a> BatchCursor<Tr2>: Cursor<Key<'a> = KC::ReadItem<'a>, ValOwn: Data, Time = Tr2::Time, Diff: Abelian>,
Bu: Builder<Time=Tr1::Time, Output = Tr2::Batch, Input: Default> + 'static,
L: FnMut(BatchKey<'_, Tr1>, &[(BatchVal<'_, Tr1>, BatchDiff<Tr1>)], &mut Vec<(BatchValOwn<Tr2>, BatchDiff<Tr2>)>)+'static,
P: FnMut(&mut Bu::Input, BatchKey<'_, Tr1>, &mut Vec<(BatchValOwn<Tr2>, Tr2::Time, BatchDiff<Tr2>)>) + 'static,
L: FnMut(KC::ReadItem<'_>, &[(BatchVal<'_, Tr1>, BatchDiff<Tr1>)], &mut Vec<(BatchValOwn<Tr2>, BatchDiff<Tr2>)>)+'static,
P: FnMut(&mut Bu::Input, KC::ReadItem<'_>, &mut Vec<(BatchValOwn<Tr2>, Tr2::Time, BatchDiff<Tr2>)>) + 'static,
{
self.reduce_core::<_,Bu,Tr2,_>(name, move |key, input, output, change| {
self.reduce_core::<_,Bu,Tr2,KC,_>(name, move |key, input, output, change| {
if !input.is_empty() {
logic(key, input, change);
}
Expand All @@ -282,17 +285,19 @@
}

/// A direct implementation of `ReduceCore::reduce_core`.
pub fn reduce_core<L, Bu, Tr2, P>(self, name: &str, logic: L, push: P) -> Arranged<'scope, TraceAgent<Tr2>>
pub fn reduce_core<L, Bu, Tr2, KC, P>(self, name: &str, logic: L, push: P) -> Arranged<'scope, TraceAgent<Tr2>>
where
Tr2: Trace<Batch: Navigable, Time=Tr1::Time>+'static,
BatchCursor<Tr1>: Cursor<Time = Tr1::Time>,
for<'a> BatchCursor<Tr2>: Cursor<Key<'a> = BatchKey<'a, Tr1>, ValOwn: Data, Time = Tr2::Time>,
KC: BatchContainer,
BatchCursor<Tr1>: Cursor<Time = Tr1::Time, KeyContainer = KC>,
for<'a> BatchCursor<Tr1>: Cursor<Key<'a> = KC::ReadItem<'a>>,
for<'a> BatchCursor<Tr2>: Cursor<Key<'a> = KC::ReadItem<'a>, ValOwn: Data, Time = Tr2::Time>,
Bu: Builder<Time=Tr1::Time, Output = Tr2::Batch, Input: Default> + 'static,
L: FnMut(BatchKey<'_, Tr1>, &[(BatchVal<'_, Tr1>, BatchDiff<Tr1>)], &mut Vec<(BatchValOwn<Tr2>, BatchDiff<Tr2>)>, &mut Vec<(BatchValOwn<Tr2>, BatchDiff<Tr2>)>)+'static,
P: FnMut(&mut Bu::Input, BatchKey<'_, Tr1>, &mut Vec<(BatchValOwn<Tr2>, Tr2::Time, BatchDiff<Tr2>)>) + 'static,
L: FnMut(KC::ReadItem<'_>, &[(BatchVal<'_, Tr1>, BatchDiff<Tr1>)], &mut Vec<(BatchValOwn<Tr2>, BatchDiff<Tr2>)>, &mut Vec<(BatchValOwn<Tr2>, BatchDiff<Tr2>)>)+'static,
P: FnMut(&mut Bu::Input, KC::ReadItem<'_>, &mut Vec<(BatchValOwn<Tr2>, Tr2::Time, BatchDiff<Tr2>)>) + 'static,
{
use crate::operators::reduce::reduce_trace;
reduce_trace::<_,Bu,_,_,_>(self, name, logic, push)
reduce_trace::<_,Bu,_,KC,_,_>(self, name, logic, push)
}
}

Expand Down
12 changes: 7 additions & 5 deletions differential-dataflow/src/operators/reduce.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -90,15 +90,17 @@
/// the value updates, as appropriate for the container. It is critical that it clear the container as
/// the operator has no ability to do this otherwise, and failing to do so represents a leak from one
/// key's computation to another, and will likely introduce non-determinism.
pub fn reduce_trace<'scope, Tr1, Bu, Tr2, L, P>(trace: Arranged<'scope, Tr1>, name: &str, logic: L, push: P) -> Arranged<'scope, TraceAgent<Tr2>>
pub fn reduce_trace<'scope, Tr1, Bu, Tr2, KC, L, P>(trace: Arranged<'scope, Tr1>, name: &str, logic: L, push: P) -> Arranged<'scope, TraceAgent<Tr2>>
where
Tr1: TraceReader<Batch: Navigable> + 'static,
Tr2: Trace<Batch: Navigable, Time = Tr1::Time> + 'static,
BatchCursor<Tr1>: Cursor<Time = Tr1::Time>,
for<'a> BatchCursor<Tr2>: Cursor<Key<'a> = BatchKey<'a, Tr1>, ValOwn: Data, Time = Tr2::Time>,
KC: BatchContainer,
BatchCursor<Tr1>: Cursor<Time = Tr1::Time, KeyContainer = KC>,
for<'a> BatchCursor<Tr1>: Cursor<Key<'a> = KC::ReadItem<'a>>,
for<'a> BatchCursor<Tr2>: Cursor<Key<'a> = KC::ReadItem<'a>, ValOwn: Data, Time = Tr2::Time>,
Bu: Builder<Time=Tr2::Time, Output = Tr2::Batch, Input: Default> + 'static,
L: FnMut(BatchKey<'_, Tr1>, &[(BatchVal<'_, Tr1>, BatchDiff<Tr1>)], &mut Vec<(BatchValOwn<Tr2>, BatchDiff<Tr2>)>, &mut Vec<(BatchValOwn<Tr2>, BatchDiff<Tr2>)>)+'static,
P: FnMut(&mut Bu::Input, BatchKey<'_, Tr1>, &mut Vec<(BatchValOwn<Tr2>, Tr2::Time, BatchDiff<Tr2>)>) + 'static,
L: FnMut(KC::ReadItem<'_>, &[(BatchVal<'_, Tr1>, BatchDiff<Tr1>)], &mut Vec<(BatchValOwn<Tr2>, BatchDiff<Tr2>)>, &mut Vec<(BatchValOwn<Tr2>, BatchDiff<Tr2>)>)+'static,
P: FnMut(&mut Bu::Input, KC::ReadItem<'_>, &mut Vec<(BatchValOwn<Tr2>, Tr2::Time, BatchDiff<Tr2>)>) + 'static,
{
reduce_with_tactic(trace, name, cursors::CursorTactic::<Tr1::Batch, Tr2::Batch, Bu, L, P>::new(logic, push))
}
Expand Down Expand Up @@ -548,13 +550,13 @@
}
}
#[inline(never)]
pub fn compute<'a, K, C1, C2, C3, L>(

Check warning on line 553 in differential-dataflow/src/operators/reduce.rs

View workflow job for this annotation

GitHub Actions / Cargo clippy 

this function has too many arguments (10/7)
&mut self,
key: K,
(source_cursor, source_storage): (&mut C1, &'a C1::Storage),
(output_cursor, output_storage): (&mut C2, &'a C2::Storage),
(batch_cursor, batch_storage): (&mut C3, &'a C3::Storage),
times: &Vec<T>,

Check warning on line 559 in differential-dataflow/src/operators/reduce.rs

View workflow job for this annotation

GitHub Actions / Cargo clippy 

writing `&Vec` instead of `&[_]` involves a new object where a slice will do
logic: &mut L,
upper_limit: &Antichain<T>,
outputs: &mut [(T, Vec<(V, T, D2)>)],
Expand Down
4 changes: 2 additions & 2 deletions differential-dataflow/tests/import.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -60,7 +60,7 @@ fn test_import_vanilla() {
::std::mem::drop(trace);
let captured =
imported
.reduce_abelian::<_,ValBuilder<_,_,_,_>,ValSpine<u64,i64,_,_>,_>(
.reduce_abelian::<_,ValBuilder<_,_,_,_>,ValSpine<u64,i64,_,_>,_,_>(
"Reduce",
|_k, s, t| t.push((s.iter().map(|&(_, w)| w).sum(), 1i64)),
|vec, key, upds| { vec.clear(); vec.extend(upds.drain(..).map(|(v,t,r)| ((key.clone(), v),t,r))); },
Expand Down Expand Up @@ -139,7 +139,7 @@ fn test_import_completed_dataflow() {
::std::mem::drop(trace);
let stream =
imported
.reduce_abelian::<_,ValBuilder<_,_,_,_>,ValSpine<u64,i64,_,_>,_>(
.reduce_abelian::<_,ValBuilder<_,_,_,_>,ValSpine<u64,i64,_,_>,_,_>(
"Reduce",
|_k, s, t| t.push((s.iter().map(|&(_, w)| w).sum(), 1i64)),
|vec, key, upds| { vec.clear(); vec.extend(upds.drain(..).map(|(v,t,r)| ((key.clone(), v),t,r)));
Expand Down
2 changes: 1 addition & 1 deletion interactive/src/backend/col.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -186,7 +186,7 @@ mod render {
if count != 0 { let mut r = Row::new(); r.push(count); output.push((r, 1)); }
}),
};
a.reduce_abelian::<_, ColValBuilder<_,_,_,_>, ColValSpine<_,_,_,_>, _>(
a.reduce_abelian::<_, ColValBuilder<_,_,_,_>, ColValSpine<_,_,_,_>, _, _>(
"Reduce",
move |k, vals, output| { f(k, vals, output); },
|col, key, upds| {
Expand Down
2 changes: 1 addition & 1 deletion interactive/src/backend/vec.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -146,7 +146,7 @@ impl Backend for VecBackend {
output.push((Value::List(items), 1));
}),
};
a.reduce_abelian::<_, ValBuilder<_, _, _, _>, ValSpine<_, _, _, _>, _>(
a.reduce_abelian::<_, ValBuilder<_, _, _, _>, ValSpine<_, _, _, _>, _, _>(
"Reduce",
move |k, v, o| f(k, v, o),
|vec, key, upds| { vec.clear(); vec.extend(upds.drain(..).map(|(v, t, r)| ((key.clone(), v), t, r))); },
Expand Down
Loading