quick-fix: Ensure schema is retained after a batching step #131
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
fpacifici
requested changes
Jun 6, 2025
Collaborator
fpacifici
left a comment
fpacifici
left a comment
There was a problem hiding this comment.
please see the comments in line
| schema=None, | ||
| ) | ||
|
|
||
| if isinstance(payload, MutableSequence) and isinstance(payload[0], tuple): |
Collaborator
There was a problem hiding this comment.
Why only a MutableSequence? Shouldn't this work with Sequences as well ?
Comment on lines
177
to
198
|
|
||
| if isinstance(payload, MutableSequence) and isinstance(payload[0], tuple): | ||
| batch = [] | ||
| schema = None | ||
| for tup in payload: | ||
| batch.append(tup[0]) | ||
|
|
||
| schema = payload[0][1] | ||
|
|
||
| msg = PyMessage( | ||
| payload=batch, | ||
| headers=[], | ||
| timestamp=timestamp, | ||
| schema=schema, | ||
| ) | ||
| else: | ||
| msg = PyMessage( | ||
| payload=payload, | ||
| headers=[], | ||
| timestamp=timestamp, | ||
| schema=None, | ||
| ) |
Collaborator
There was a problem hiding this comment.
Please let's write unit tests for this logic (there is already a test_reduce_delegate you can add onto).
As this is a bit of a hack, a good way to keep it under control is to have this behavior well tested so that when we refactor this we will not miss corner cases.
Contributor
Author
There was a problem hiding this comment.
will do. Initially I just wanted to check if we're okay with this sort of hack in the code.
Contributor
Author
There was a problem hiding this comment.
This test is already checking what we need here. I made it a bit more "complex"
ayirr7
changed the title
fpacifici
approved these changes
Jun 11, 2025
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.
A really simple way to unblock the work where a batching step happens before parsing. I can also introduce the concept of
Schemain this PR, where we have an enum of all the allowed schemas / topic names.We should not stick with this in the long term, and likely enforce all Accumulators/Reduce steps to output a
Tuple[MutableSequence[InputType], Schema]. Making this is a more significant typing change as it involves making sure primitives likeFlatMapare also able to handle that output type.