[SPARK-46064][SQL][SS] Move out EliminateEventTimeWatermark to the analyzer and change to only take effect on resolved child #43971
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
…r and change to only take effect on resolved child
HeartSaVioR
changed the title
|
cc. @cloud-fan PTAL, thanks! |
HeartSaVioR
commented
Nov 23, 2023
| @@ -347,7 +347,9 @@ class Analyzer(override val catalogManager: CatalogManager) extends RuleExecutor | |||
| Batch("Cleanup", fixedPoint, | |||
| CleanupAliases), | |||
| Batch("HandleSpecialCommand", Once, | |||
| HandleSpecialCommand) | |||
| HandleSpecialCommand), | |||
| Batch("Remove watermark for batch query", Once, | |||
There was a problem hiding this comment.
I guess the node is expected to be already resolved from this point, but I'm checking the flag of resolved in EliminateEventTimeWatermark (again), to make the logic be self-describe.
HeartSaVioR
commented
Nov 23, 2023
| @@ -68,4 +68,18 @@ object TestRelations { | |||
|
|
|||
| val mapRelation = LocalRelation( | |||
| AttributeReference("map", MapType(IntegerType, IntegerType))()) | |||
|
|
|||
| val streamingRelation = LocalRelation( | |||
There was a problem hiding this comment.
This is also added in #43966 - I'll rebase when either one is merged.
cloud-fan
approved these changes
Nov 23, 2023
… to be eliminated
|
https://github.com/HeartSaVioR/spark/actions/runs/6967344462/job/18959281391 failures do not relate to the change. SQL tests are all passing. |
|
Thanks, merging to master/3.5/3.4. |
HeartSaVioR
added a commit
that referenced
this pull request
Nov 23, 2023
…alyzer and change to only take effect on resolved child This PR proposes to move out EliminateEventTimeWatermark to the analyzer (one of the analysis rule), and also make a change to eliminate EventTimeWatermark node only when the child of EventTimeWatermark is "resolved". Currently, we apply EliminateEventTimeWatermark immediately when withWatermark is called, which means the rule is applied immediately against the child, regardless whether child is resolved or not. It is not an issue for the usage of DataFrame API initiated by read / readStream, because streaming sources have the flag isStreaming set to true even it is yet resolved. But mix-up of SQL and DataFrame API would expose the issue; we may not know the exact value of isStreaming flag on unresolved node and it is subject to change upon resolution. No. New UTs. No. Closes #43971 from HeartSaVioR/SPARK-46064. Authored-by: Jungtaek Lim <kabhwan.opensource@gmail.com> Signed-off-by: Jungtaek Lim <kabhwan.opensource@gmail.com> (cherry picked from commit a703dac) Signed-off-by: Jungtaek Lim <kabhwan.opensource@gmail.com>
HeartSaVioR
added a commit
that referenced
this pull request
Nov 23, 2023
…alyzer and change to only take effect on resolved child This PR proposes to move out EliminateEventTimeWatermark to the analyzer (one of the analysis rule), and also make a change to eliminate EventTimeWatermark node only when the child of EventTimeWatermark is "resolved". Currently, we apply EliminateEventTimeWatermark immediately when withWatermark is called, which means the rule is applied immediately against the child, regardless whether child is resolved or not. It is not an issue for the usage of DataFrame API initiated by read / readStream, because streaming sources have the flag isStreaming set to true even it is yet resolved. But mix-up of SQL and DataFrame API would expose the issue; we may not know the exact value of isStreaming flag on unresolved node and it is subject to change upon resolution. No. New UTs. No. Closes #43971 from HeartSaVioR/SPARK-46064. Authored-by: Jungtaek Lim <kabhwan.opensource@gmail.com> Signed-off-by: Jungtaek Lim <kabhwan.opensource@gmail.com> (cherry picked from commit a703dac) Signed-off-by: Jungtaek Lim <kabhwan.opensource@gmail.com>
szehon-ho
pushed a commit
to szehon-ho/spark
that referenced
this pull request
Feb 7, 2024
…alyzer and change to only take effect on resolved child This PR proposes to move out EliminateEventTimeWatermark to the analyzer (one of the analysis rule), and also make a change to eliminate EventTimeWatermark node only when the child of EventTimeWatermark is "resolved". Currently, we apply EliminateEventTimeWatermark immediately when withWatermark is called, which means the rule is applied immediately against the child, regardless whether child is resolved or not. It is not an issue for the usage of DataFrame API initiated by read / readStream, because streaming sources have the flag isStreaming set to true even it is yet resolved. But mix-up of SQL and DataFrame API would expose the issue; we may not know the exact value of isStreaming flag on unresolved node and it is subject to change upon resolution. No. New UTs. No. Closes apache#43971 from HeartSaVioR/SPARK-46064. Authored-by: Jungtaek Lim <kabhwan.opensource@gmail.com> Signed-off-by: Jungtaek Lim <kabhwan.opensource@gmail.com> (cherry picked from commit a703dac) Signed-off-by: Jungtaek Lim <kabhwan.opensource@gmail.com>
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?
This PR proposes to move out EliminateEventTimeWatermark to the analyzer (one of the analysis rule), and also make a change to eliminate EventTimeWatermark node only when the child of EventTimeWatermark is "resolved".
Why are the changes needed?
Currently, we apply EliminateEventTimeWatermark immediately when withWatermark is called, which means the rule is applied immediately against the child, regardless whether child is resolved or not.
It is not an issue for the usage of DataFrame API initiated by read / readStream, because streaming sources have the flag isStreaming set to true even it is yet resolved. But mix-up of SQL and DataFrame API would expose the issue; we may not know the exact value of isStreaming flag on unresolved node and it is subject to change upon resolution.
Does this PR introduce any user-facing change?
No.
How was this patch tested?
New UTs.
Was this patch authored or co-authored using generative AI tooling?
No.