Merge batcher input generic over containers #494

antiguru · 2024-05-23T20:31:08Z

Add infrastructure and implementations to support containers, both specific and general, from timely streams to merge batchers. This is the last piece to enable building arrangements from non-vector input data.

The change adds several new traits and implementations. The changes fall into roughly two categories:

Capture behavior to transcribe streams of containers to chunks of sorted and consolidated data, suitable for integration in merge batchers.
Split of merge batchers into chunk formation and chain maintenance.

The first allows spines to express more opinions on how streams of containers can be transformed into chunks, either by supplying specialized implementations that know exactly how to map specific containers, or by instructing a generic implementation with specific information about containers.

The second decouples the merge batcher's chain maintaining from forming chunks, which seems the right thing to do given that the two are distinct tasks.

frankmcsherry

Many comments, but marking "approve" for now because it seems safe to land if not yet perfect. Many places would love more comments, which you already warned me about. A few other notes here and there.

I think not too far off I might try and write some mod comments to try and make sure I can stay on top of all of the concepts.

src/consolidation.rs

frankmcsherry · 2024-06-04T00:07:33Z

src/consolidation.rs

+/// Layout of data to be consolidated.
+// TODO: This could be split in two, to separate sorting and consolidation.
+pub trait ConsolidateLayout: Container {


I know we talked about this, but I now have no clue what it does. :D

src/consolidation.rs

src/trace/implementations/merge_batcher_flat.rs

frankmcsherry · 2024-06-04T00:54:24Z

src/trace/implementations/merge_batcher_flat.rs

+        if result.len() > 0 {
+            output.push(result);
+        } else {
+            self.recycle(result, stash);
+        }


This seems weird! At least, you are about to copy more things into result, but push out a potentially non-full vector. Is it on purpose?

src/trace/implementations/merge_batcher_flat.rs

Signed-off-by: Moritz Hoffmann <[email protected]>

Make the BuilderInput trait generic over the key and value containers instead of the layout, which makes trait bounds much simpler and allows for reusing the `BuilerInput` implementation for generic key and value containers, instead tying it to specific layouts. Signed-off-by: Moritz Hoffmann <[email protected]>

Signed-off-by: Moritz Hoffmann <[email protected]>

antiguru · 2024-06-13T17:55:46Z

src/trace/implementations/chunker.rs

+pub struct VecChunker<T> {
+    pending: Vec<T>,
+    ready: VecDeque<Vec<T>>,
+    empty: Option<Vec<T>>,
+}


Could unify the type with ColumnationChunker below if that's helpful.

frankmcsherry

Looks good. Some comments around whether some chunker types are truly necessary. If they are not, no worries, and we can try and figure out off the critical path if they can be optimized out.

src/consolidation.rs

frankmcsherry · 2024-06-13T18:11:24Z

src/trace/implementations/chunker.rs

+use crate::consolidation::{consolidate_updates, ConsolidateContainer};
+use crate::difference::Semigroup;
+
+/// Chunk a stream of vectors into chains of vectors.


Is this materially different from a capacity container builder? If they are very similar, having one fewer abstraction is great! If there is a really important distinction, we should write it down so that I don't accidentally optimize this away in the future. :D

Edit: perhaps rather compared to consolidating container builder, as it doesn't just choose a capacity.

Not very different, but somewhat incompatible! The consolidating container builder has a PushInto implementation for individual items, but here we want to push whole containers (RefOrMut<...>). Since the other implementation already exists, we cannot define a impl for just RefOrMut because that would conflict.

There is a path out of this mess, but I don't know if we want to go down that route: We could introduce a ContainerMarker<C>(C) struct that we can use to specialize the PushInto implementation for the consolidating container builder. Since nothing that's further up in the hierarchy can know this type, Rust knows that there can't be conflicting implementations. If we had that, we might be able to remove the VecChunker.

That aside, having $n$ different implementations that are roughly the same isn't great, and I believe with a bit more effort we could tidy it up.

frankmcsherry · 2024-06-13T18:17:05Z

src/trace/implementations/merge_batcher.rs

 where
-    M: Merger,
+    C: ContainerBuilder<Container=M::Chunk> + Default,


Is Input really necessary as part of the struct, or can the PushInto<Input> constraint be added somewhere? Alternately, could it be added here to signpost what it is and why it exists (or just doc it).

In principle, it can be removed. Then we'd only carry a PushInto bound for C with us and everything should work. But .. for flatcontainers, Rust has a hard/impossible time to figure out what the scope's timestamp type would be and I couldn't figure out how to convince it to carry that information. Maybe it's a matter of trying a bit harder, but I feel we can do that outside of this change?

Signed-off-by: Moritz Hoffmann <[email protected]>

antiguru · 2024-06-13T20:04:29Z

tests/bfs.rs

-#[test] fn bfs_10_20_1000() { test_sizes(10, 20, 1000, Config::process(3)); }
-#[test] fn bfs_100_200_10() { test_sizes(100, 200, 10, Config::process(3)); }
-#[test] fn bfs_100_2000_1() { test_sizes(100, 2000, 1, Config::process(3)); }
+#[test] fn bfs_10_20_1000() { test_sizes(10, 20, 1000, 3); }


Config isn't Clone...

antiguru force-pushed the flatcontainer_e2e branch 5 times, most recently from 6d52449 to 2520e6f Compare May 31, 2024 18:41

antiguru force-pushed the flatcontainer_e2e branch 3 times, most recently from 0daf37c to 4d350c7 Compare June 4, 2024 00:53

frankmcsherry approved these changes Jun 4, 2024

View reviewed changes

antiguru changed the title ~~Merge batcher for flatcontainer~~ Merge batcher input generic over containers Jun 4, 2024

antiguru mentioned this pull request Jun 4, 2024

Implement advance_by for FlatStack's iterator antiguru/flatcontainer#36

Open

antiguru force-pushed the flatcontainer_e2e branch from 21d4e21 to 8f4cf39 Compare June 4, 2024 19:04

antiguru marked this pull request as ready for review June 4, 2024 19:06

antiguru force-pushed the flatcontainer_e2e branch 2 times, most recently from c7c4aa8 to c29adf6 Compare June 12, 2024 14:20

antiguru added 11 commits June 13, 2024 11:47

Generic chunker, plus flatcontainer support

d42d6a6

Signed-off-by: Moritz Hoffmann <[email protected]>

Address comments

4f96250

Signed-off-by: Moritz Hoffmann <[email protected]>

Make chunker public

cf44db5

Signed-off-by: Moritz Hoffmann <[email protected]>

Unify sorting and consolidation

83a4f9e

Signed-off-by: Moritz Hoffmann <[email protected]>

Add comments.

94cfb56

Signed-off-by: Moritz Hoffmann <[email protected]>

Address comments

10d5f62

Signed-off-by: Moritz Hoffmann <[email protected]>

BFS for flatcontainer

c7c4131

Signed-off-by: Moritz Hoffmann <[email protected]>

Fix merge batcher flat

31efb51

Signed-off-by: Moritz Hoffmann <[email protected]>

Update flatcontainer test to use flatcontainers in iteration

f52db45

Signed-off-by: Moritz Hoffmann <[email protected]>

Replace Chunker with container builder

e279663

Signed-off-by: Moritz Hoffmann <[email protected]>

antiguru force-pushed the flatcontainer_e2e branch from c29adf6 to e279663 Compare June 13, 2024 15:54

Remove ContainerChunker input parameter

08aeb54

Signed-off-by: Moritz Hoffmann <[email protected]>

antiguru commented Jun 13, 2024

View reviewed changes

frankmcsherry approved these changes Jun 13, 2024

View reviewed changes

antiguru added 2 commits June 13, 2024 15:45

Free consolidate_container function

6bbf7a5

Signed-off-by: Moritz Hoffmann <[email protected]>

Integrate flatcontainer bfs in regular bfs

7a70f88

Signed-off-by: Moritz Hoffmann <[email protected]>

antiguru commented Jun 13, 2024

View reviewed changes

antiguru merged commit d0a1cab into TimelyDataflow:master Jun 13, 2024
7 checks passed

antiguru deleted the flatcontainer_e2e branch June 13, 2024 20:06

antiguru mentioned this pull request Jun 13, 2024

Introduce generic consolidator #480

Closed

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

Merge batcher input generic over containers #494

Merge batcher input generic over containers #494

antiguru commented May 23, 2024 •

edited

Loading

frankmcsherry left a comment

frankmcsherry Jun 4, 2024

frankmcsherry Jun 4, 2024

antiguru Jun 13, 2024

frankmcsherry left a comment

frankmcsherry Jun 13, 2024

frankmcsherry Jun 13, 2024

antiguru Jun 13, 2024

frankmcsherry Jun 13, 2024

antiguru Jun 13, 2024

antiguru Jun 13, 2024

Merge batcher input generic over containers #494

Merge batcher input generic over containers #494

Conversation

antiguru commented May 23, 2024 • edited Loading

frankmcsherry left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

frankmcsherry left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

antiguru commented May 23, 2024 •

edited

Loading