Join vertex - temporary PR on my own fork #1
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
Signed-off-by: Julie Vogelmani <[email protected]>
Signed-off-by: Julie Vogelmani <[email protected]>
Signed-off-by: Julie Vogelmani <[email protected]>
Signed-off-by: Julie Vogelmani <[email protected]>
Signed-off-by: Julie Vogelmani <[email protected]>
Signed-off-by: Julie Vogelmani <[email protected]>
Signed-off-by: Julie Vogelmani <[email protected]>
Signed-off-by: Julie Vogelmani <[email protected]>
Signed-off-by: Julie Vogelmani <[email protected]>
Signed-off-by: Julie Vogelmani <[email protected]>
Signed-off-by: Julie Vogelmani <[email protected]>
Signed-off-by: Julie Vogelmani <[email protected]>
Signed-off-by: Julie Vogelmani <[email protected]>
Signed-off-by: Julie Vogelmani <[email protected]>
Signed-off-by: Julie Vogelmani <[email protected]>
…ishing to the same buffer Signed-off-by: Julie Vogelmani <[email protected]>
Signed-off-by: Julie Vogelmani <[email protected]>
Signed-off-by: Julie Vogelmani <[email protected]>
Signed-off-by: Julie Vogelmani <[email protected]>
Signed-off-by: Julie Vogelmani <[email protected]>
Signed-off-by: Julie Vogelmani <[email protected]>
Signed-off-by: Julie Vogelmani <[email protected]>
Signed-off-by: Julie Vogelmani <[email protected]>
Signed-off-by: Julie Vogelmani <[email protected]>
Signed-off-by: Julie Vogelmani <[email protected]>
Signed-off-by: Julie Vogelmani <[email protected]>
Signed-off-by: Julie Vogelmani <[email protected]>
Signed-off-by: Julie Vogelmani <[email protected]>
Signed-off-by: Julie Vogelman <[email protected]>
Signed-off-by: Julie Vogelman <[email protected]>
Signed-off-by: Julie Vogelman <[email protected]>
Signed-off-by: Julie Vogelman <[email protected]>
Signed-off-by: Julie Vogelman <[email protected]>
Signed-off-by: Julie Vogelman <[email protected]>
Signed-off-by: Julie Vogelman <[email protected]>
Signed-off-by: Julie Vogelman <[email protected]>
Signed-off-by: Julie Vogelman <[email protected]>
Signed-off-by: Julie Vogelman <[email protected]>
Signed-off-by: Julie Vogelman <[email protected]>
Signed-off-by: Julie Vogelman <[email protected]>
Signed-off-by: Julie Vogelman <[email protected]>
Signed-off-by: Julie Vogelman <[email protected]>
Signed-off-by: Julie Vogelman <[email protected]>
Signed-off-by: Julie Vogelman <[email protected]>
Signed-off-by: Julie Vogelman <[email protected]>
Signed-off-by: Julie Vogelman <[email protected]>
juliev0
commented
Jul 18, 2023
| @@ -261,7 +261,7 @@ func Test_batchSyncWithMaxBatchSize(t *testing.T) { | |||
| assert.NoError(t, err) | |||
| err = wal.Write(&message) | |||
| assert.NoError(t, err) | |||
| assert.Equal(t, int64(222), tempWAL.prevSyncedWOffset) | |||
| assert.Equal(t, int64(252), tempWAL.prevSyncedWOffset) | |||
There was a problem hiding this comment.
because the ID changed to include vertex name, partition, etc, it made the overall message longer.
juliev0
commented
Jul 18, 2023
| @@ -137,7 +137,7 @@ func (sp *SourceProcessor) Start(ctx context.Context) error { | |||
| shuffleFuncMap := make(map[string]*shuffle.Shuffle) | |||
| for _, edge := range sp.VertexInstance.Vertex.Spec.ToEdges { | |||
| if edge.ToVertexType == dfv1.VertexTypeReduceUDF && edge.GetToVertexPartitionCount() > 1 { | |||
| s := shuffle.NewShuffle(sp.VertexInstance.Vertex.GetName(), edge.GetToVertexPartitionCount()) | |||
| s := shuffle.NewShuffle(edge.To, edge.GetToVertexPartitionCount()) | |||
There was a problem hiding this comment.
All Vertices publishing to a given Reduce Vertex need to be using the same hashing seed for distributing messages - thus, use the receiving Vertex name for that
Signed-off-by: Julie Vogelman <[email protected]>
juliev0
commented
Jul 18, 2023
| // we only consider idle watermark if it is smaller than or equal to min of all the last processed watermarks. | ||
| if headWMB.Watermark > e.getMinFromLastProcessed(headWMB.Watermark) { | ||
| return wmb.WMB{} | ||
| } |
There was a problem hiding this comment.
this needed to be moved into EdgeFetcherSet.GetHeadWMB()
Signed-off-by: Julie Vogelman <[email protected]>
juliev0
commented
Jul 18, 2023
| @@ -67,7 +67,7 @@ metadata: | |||
| spec: | |||
| redis: | |||
| native: | |||
| version: 6.2.6` | |||
| version: 7.0.11` | |||
There was a problem hiding this comment.
maybe should move this into a separate PR - just noticed it was outdated
juliev0
commented
Jul 18, 2023
| containerTemplate: | ||
| env: | ||
| - name: NUMAFLOW_DEBUG | ||
| value: "true" |
There was a problem hiding this comment.
maybe should take out debug level environment variable for the sake of parity with the other tests
Signed-off-by: Julie Vogelman <[email protected]>
Signed-off-by: Julie Vogelman <[email protected]>
Signed-off-by: Julie Vogelman <[email protected]>
Signed-off-by: Julie Vogelman <[email protected]>
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.
Fixes numaproj#719
Enables Join Vertex (multiple Vertices publish to a single Vertex). This includes both cases where the Join Vertex is a map type, sink type, or Reduce type.
The implementation of this is based on this design document. (Sorry it's an internal link for now.)
Changes
The Design document goes over the need for an EdgeFetcherSet, which is the set of
EdgeFetchersthat a Vertex consumes Watermarks from.EdgeFetcherSettraverses theEdgeFetchersfor each incoming Edge.In addition to that, some other changes made include:
Cycles
Plan is to enable Cycles in the pipeline as well as long as there is no Reduce Vertex at the point of the cycle or to the right (since Reduce vertex relies on the Watermark, and Cycles create a scenario in which late data arrives after Close-of-book). As of yet, Cycles have not been tested, and also validation needs to be added to prevent the Cycles mentioned above.