Container GATs, improve traits #541
Merged
Conversation
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
antiguru
force-pushed
the
container_gat
branch
from
15e1740
to
bb94b15
Compare
Signed-off-by: Moritz Hoffmann <[email protected]>
antiguru
force-pushed
the
container_gat
branch
from
4801f8d
to
3cefd09
Compare
antiguru
changed the title
|
I reworked the PR to only contain a minimum of changes. This compiles with differential once applying the following patch: diff --git a/src/collection.rs b/src/collection.rs
index f2daf8ad..20cab0d2 100644
--- a/src/collection.rs
+++ b/src/collection.rs
@@ -416,10 +416,10 @@ impl<G: Scope, D: Data, R: Semigroup> Collection<G, D, R> where G::Timestamp: Da
/// .inspect_batch(|t,xs| println!("errors @ {:?}: {:?}", t, xs));
/// });
/// ```
- pub fn inspect_batch<F>(&self, func: F) -> Collection<G, D, R>
+ pub fn inspect_batch<F>(&self, mut func: F) -> Collection<G, D, R>
where F: FnMut(&G::Timestamp, &[(D, G::Timestamp, R)])+'static {
self.inner
- .inspect_batch(func)
+ .inspect_batch(move |time, batch| func(time, &batch[..]))
.as_collection()
}
/// Attaches a timely dataflow probe to the output of a Collection.
( |
antiguru
force-pushed
the
container_gat
branch
2 times, most recently
from
ac6e383
to
f117d82
Compare
Containers are generic over their contents, and over reading and draining. Signed-off-by: Moritz Hoffmann <[email protected]>
Signed-off-by: Moritz Hoffmann <[email protected]>
frankmcsherry
approved these changes
Mar 19, 2024
Comment on lines
+30
to
+31
| /// of all pieces must be equal to the length of the original container. When combining | ||
| /// containers, the length of the result must be the sum of the individual parts. |
There was a problem hiding this comment.
At this point we do not know what it means to "combine a container".
| type DrainIter<'a>: Iterator<Item=Self::Item<'a>>; | ||
|
|
||
| /// Returns an iterator that drains the contents of this container. | ||
| /// Drain leaves the container in an undefined state. |
There was a problem hiding this comment.
We could/should say more about what "undefined state" means. Perhaps "it is an error to do anything other than clear or drop" or " unless you are certain the type is FOO".
| F: FnMut(usize, &mut Self); | ||
| } | ||
|
|
||
| impl<T: Clone + 'static> PushPartitioned for Vec<T> { | ||
| impl<T: PushContainer + 'static> PushPartitioned for T where for<'a> T::Item<'a>: PushInto<T> { |
There was a problem hiding this comment.
Discussion: might we want to make push_partitioned into a free function generic in a T1: Container and T2: PushContainer, reading from the first and writing in to the second?
| @@ -7,7 +7,7 @@ use crate::dataflow::operators::generic::operator::Operator; | |||
| use crate::dataflow::{Scope, StreamCore}; | |||
|
|
|||
| /// Exchange records between workers. | |||
| pub trait Exchange<D> { | |||
| pub trait Exchange<C: PushPartitioned> { | |||
There was a problem hiding this comment.
From discussion: This could be C: Container because it is only used to bind C::Item<'a>.
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
Add this suggestion to a batch that can be applied as a single commit.
This suggestion is invalid because no changes were made to the code.
Suggestions cannot be applied while the pull request is closed.
Suggestions cannot be applied while viewing a subset of changes.
Only one suggestion per line can be applied in a batch.
Add this suggestion to a batch that can be applied as a single commit.
Applying suggestions on deleted lines is not supported.
You must change the existing code in this line in order to create a valid suggestion.
Outdated suggestions cannot be applied.
This suggestion has been applied or marked resolved.
Suggestions cannot be applied from pending reviews.
Suggestions cannot be applied on multi-line comments.
Suggestions cannot be applied while the pull request is queued to merge.
Suggestion cannot be applied right now. Please check back later.
Cleanup the
Container::Itemtype, move it toPushPartitionedand make it a GAT. Similarly, split theMaptrait in two (Map,MapInPlace) and support GATs in theMaptrait.This has the benefit that Timely becomes less opinionated about the interface provided by containers, for example by allowing non-owned data when calling
stream.map.