MapOperator and JoinOperator Implementation. #85
Merged
Conversation
This file contains hidden or 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
Collaborator
- This PR provides the implementation of MapOperator and JoinOperator along with the relevant tests.
Previous code allowed either OperatorConfig / arguments that could be used to construct OperatorConfig to be passed to connect_*_write. However, Rust wasn't able to infer types if the passed argument was a parameterized implementation of a generic type. This fix forces connects to take an OperatorConfig.
This commit adds a MapOperator that takes in an input stream of type D1 and writes to an output stream of type D2. The user provides a function to convert the messages from type D1 to D2 using the OperatorConfig.
This commit adds the JoinOperator implementation, its tests and introduces a dependency to the chashmap crate. The main.rs became incompatible with the API of the JoinOperator and was changed.
The previous implementation used per-state watermark callbacks (that had mutable access to the stream state) to clean up the state after the join callback is invoked. However, this was buggy, since consecutive watermark callbacks could clear up state that might be required for the join callback (when one stream is lagging behind the other). This fix cleans up state after the join callback is called.
This commit uses an ExtractStream to assert the correctness of the MapOperator and the JoinOperator. The previous ReceiveOperator ran callbacks in a different thread and would not fail tests if provided with incorrect implementations.
This commit fixes the tests and adds the Operator trait for the Map and the JoinOperator.
pschafhalter
requested changes
Mar 2, 2020
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.