[SPARK-22187][SS] Update unsaferow format for saved state in flatMapGroupsWithState to allow timeouts with deleted state #21739
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
tdas
commented
Jul 9, 2018
| @@ -43,7 +43,7 @@ case class ObjectType(cls: Class[_]) extends DataType { | |||
|
|
|||
| def asNullable: DataType = this | |||
|
|
|||
| override def simpleString: String = cls.getName | |||
| override def simpleString: String = s"Object[${cls.getName}]" | |||
There was a problem hiding this comment.
this is to make it differentiate between IntegerType and ObjectType with int in it. Both show up in the simpleStreaing as int.
|
Test build #92766 has finished for PR 21739 at commit
|
|
Test build #92790 has finished for PR 21739 at commit
|
tdas
changed the title
|
jenkins retest this please |
|
Test build #92836 has finished for PR 21739 at commit
|
tdas
commented
Jul 11, 2018
| } | ||
|
|
||
|
|
||
| private class StateManagerImplV2( |
There was a problem hiding this comment.
Add docs explaining the state format
|
Test build #92853 has finished for PR 21739 at commit
|
|
Test build #92845 has finished for PR 21739 at commit
|
|
Test build #92846 has finished for PR 21739 at commit
|
|
LGTM |
|
Test build #93258 has finished for PR 21739 at commit
|
|
LGTM again |
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.
What changes were proposed in this pull request?
Currently, the group state of user-defined-type is encoded as top-level columns in the UnsafeRows stores in the state store. The timeout timestamp is also saved as (when needed) as the last top-level column. Since the group state is serialized to top-level columns, you cannot save "null" as a value of state (setting null in all the top-level columns is not equivalent). So we don't let the user set the timeout without initializing the state for a key. Based on user experience, this leads to confusion.
This PR is to change the row format such that the state is saved as nested columns. This would allow the state to be set to null, and avoid these confusing corner cases. However, queries recovering from existing checkpoint will use the previous format to maintain compatibility with existing production queries.
How was this patch tested?
Refactored existing end-to-end tests and added new tests for explicitly testing obj-to-row conversion for both state formats.