You are viewing a plain text version of this content. The canonical link for it is here.
Posted to reviews@spark.apache.org by GitBox <gi...@apache.org> on 2022/02/28 12:46:48 UTC
[GitHub] [spark] nyingping opened a new pull request #35680: Spark 38349
nyingping opened a new pull request #35680:
URL: https://github.com/apache/spark/pull/35680
<!--
Thanks for sending a pull request! Here are some tips for you:
1. If this is your first time, please read our contributor guidelines: https://spark.apache.org/contributing.html
2. Ensure you have added or run the appropriate tests for your PR: https://spark.apache.org/developer-tools.html
3. If the PR is unfinished, add '[WIP]' in your PR title, e.g., '[WIP][SPARK-XXXX] Your PR title ...'.
4. Be sure to keep the PR description updated to reflect all changes.
5. Please write your PR title to summarize what this PR proposes.
6. If possible, provide a concise example to reproduce the issue for a faster review.
7. If you want to add a new configuration, please read the guideline first for naming configurations in
'core/src/main/scala/org/apache/spark/internal/config/ConfigEntry.scala'.
8. If you want to add or modify an error type or message, please read the guideline first in
'core/src/main/resources/error/README.md'.
-->
### What changes were proposed in this pull request?
<!--
Please clarify what changes you are proposing. The purpose of this section is to outline the changes and how this PR fixes the issue.
If possible, please consider writing useful notes for better and faster reviews in your PR. See the examples below.
1. If you refactor some codes with changing classes, showing the class hierarchy will help reviewers.
2. If you fix some SQL features, you can provide some references of other DBMSes.
3. If there is design documentation, please add the link.
4. If there is a discussion in the mailing list, please add the link.
-->
No need to filter events when sessionwindow gapDuration greater than 0
### Why are the changes needed?
<!--
Please clarify why the changes are needed. For instance,
1. If you propose a new API, clarify the use case for a new API.
2. If you fix a bug, you can clarify why it is a bug.
-->
save calculation resources and improve performance.
### Does this PR introduce _any_ user-facing change?
<!--
Note that it means *any* user-facing change including all aspects such as the documentation fix.
If yes, please clarify the previous behavior and the change this PR proposes - provide the console output, description and/or an example to show the behavior difference if possible.
If possible, please also clarify if this is a user-facing change compared to the released Spark versions or within the unreleased branches such as master.
If no, write 'No'.
-->
No
### How was this patch tested?
<!--
If tests were added, say they were added here. Please make sure to add some test cases that check the changes thoroughly including negative and positive cases if possible.
If it was tested in a way different from regular unit tests, please clarify how you tested step by step, ideally copy and paste-able, so that other reviewers can test and check, and descendants can verify in the future.
If tests were not added, please describe why they were not added and/or why it was difficult to add.
If benchmark tests were added, please run the benchmarks in GitHub Actions for the consistent environment, and the instructions could accord to: https://spark.apache.org/developer-tools.html#github-workflow-benchmarks.
-->
add new UT and simple benchmark.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] nyingping commented on pull request #35680: [SPARK-38349][SS] No need to filter events when sessionwindow gapDuration greater than 0
Posted by GitBox <gi...@apache.org>.
nyingping commented on pull request #35680:
URL: https://github.com/apache/spark/pull/35680#issuecomment-1082577852
@HeartSaVioR @viirya Thanks a lot!
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] HeartSaVioR commented on pull request #35680: [SPARK-38349][SS] No need to filter events when sessionwindow gapDuration greater than 0
Posted by GitBox <gi...@apache.org>.
HeartSaVioR commented on pull request #35680:
URL: https://github.com/apache/spark/pull/35680#issuecomment-1076315817
Sorry I have been very busy with handling stuff. I see the rationalization, but it is a bit unclear which cases we want to optimize (the case gapDuration produces > 0 consistently), and the code change does not seem to be intuitive given that the code tries to disassemble with Cast.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] AmplabJenkins commented on pull request #35680: [SPARK-38349][SS]No need to filter events when sessionwindow gapDuration greater than 0
Posted by GitBox <gi...@apache.org>.
AmplabJenkins commented on pull request #35680:
URL: https://github.com/apache/spark/pull/35680#issuecomment-1054924888
Can one of the admins verify this patch?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] HeartSaVioR edited a comment on pull request #35680: [SPARK-38349][SS] No need to filter events when sessionwindow gapDuration greater than 0
Posted by GitBox <gi...@apache.org>.
HeartSaVioR edited a comment on pull request #35680:
URL: https://github.com/apache/spark/pull/35680#issuecomment-1076315817
Sorry I have been very busy with handling stuff. I see the rationalization, but it is a bit unclear in the code which cases we want to optimize (the case gapDuration produces > 0 consistently), and the code change does not seem to be intuitive given that the code tries to disassemble with Cast.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] HeartSaVioR commented on pull request #35680: [SPARK-38349][SS] No need to filter events when sessionwindow gapDuration greater than 0
Posted by GitBox <gi...@apache.org>.
HeartSaVioR commented on pull request #35680:
URL: https://github.com/apache/spark/pull/35680#issuecomment-1082570576
Thanks @nyingping for your contribution! I merged this to master.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] nyingping commented on pull request #35680: [SPARK-38214][SS]No need to filter events when sessionwindow gapDuration greater than 0
Posted by GitBox <gi...@apache.org>.
nyingping commented on pull request #35680:
URL: https://github.com/apache/spark/pull/35680#issuecomment-1054250714
I will check it later.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] HeartSaVioR commented on a change in pull request #35680: [SPARK-38349][SS] No need to filter events when sessionwindow gapDuration greater than 0
Posted by GitBox <gi...@apache.org>.
HeartSaVioR commented on a change in pull request #35680:
URL: https://github.com/apache/spark/pull/35680#discussion_r836229369
##########
File path: sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/Analyzer.scala
##########
@@ -4061,10 +4061,23 @@ object SessionWindowing extends Rule[LogicalPlan] {
case s: SessionWindow => sessionAttr
}
+ val filterTimeSize = gapDuration.child.dataType match {
Review comment:
I guess we can simplify this a bit more like below:
```
val filterByTimeRange = session.gapDuration match {
case Literal(interval: CalendarInterval, CalendarIntervalType) =>
interval == null || interval.months + interval.days + interval.microseconds <= 0
case _ => true
}
```
##########
File path: sql/core/src/test/scala/org/apache/spark/sql/DataFrameSessionWindowingSuite.scala
##########
@@ -495,4 +495,65 @@ class DataFrameSessionWindowingSuite extends QueryTest with SharedSparkSession
validateWindowColumnInSchema(schema2, "session")
}
}
+
+ test("SPARK-38349: No need to filter events when gapDuration greater than 0") {
+ // negative value
Review comment:
Logic for verification can be deduplicated via adding inner-method in this test like below:
```
def checkFilterCondition(
logicalPlan: LogicalPlan,
expectTimeRange: Boolean,
assertHintMsg: String): Unit = {
val filter = logicalPlan.find { plan =>
plan.isInstanceOf[Filter] && plan.children.head.isInstanceOf[Project]
}
assert(filter.isDefined)
val exist = filter.get.expressions.flatMap { expr =>
expr.collect { case gt: GreaterThan => gt }
}
if (expectTimeRange) {
assert(exist.nonEmpty, assertHintMsg)
} else {
assert(exist.isEmpty, assertHintMsg)
}
}
```
and call the method per case.
Let's use logicalPlan (`df.queryExecution.logical`) instead of optimizedPlan which Spark can optimize the plan and may have changed the plan out of the expectation.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] nyingping commented on pull request #35680: [SPARK-38349][SS] No need to filter events when sessionwindow gapDuration greater than 0
Posted by GitBox <gi...@apache.org>.
nyingping commented on pull request #35680:
URL: https://github.com/apache/spark/pull/35680#issuecomment-1077383248
sure,I think it is acceptable that only deal with static gaps will greatly reduce uncertainty.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] viirya commented on a change in pull request #35680: [SPARK-38349][SS] No need to filter events when sessionwindow gapDuration greater than 0
Posted by GitBox <gi...@apache.org>.
viirya commented on a change in pull request #35680:
URL: https://github.com/apache/spark/pull/35680#discussion_r833827343
##########
File path: sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/Analyzer.scala
##########
@@ -4061,10 +4061,30 @@ object SessionWindowing extends Rule[LogicalPlan] {
case s: SessionWindow => sessionAttr
}
+ def isGapInvalid(child: Expression): Boolean = {
+ val calendarInterval = IntervalUtils
+ .safeStringToInterval(UTF8String.fromString(child.toString))
Review comment:
I think this only applies for `Literal` and we need to evaluate it, instead of calling `toString`.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] viirya commented on a change in pull request #35680: [SPARK-38349][SS] No need to filter events when sessionwindow gapDuration greater than 0
Posted by GitBox <gi...@apache.org>.
viirya commented on a change in pull request #35680:
URL: https://github.com/apache/spark/pull/35680#discussion_r836810791
##########
File path: sql/core/src/test/scala/org/apache/spark/sql/DataFrameSessionWindowingSuite.scala
##########
@@ -495,4 +495,65 @@ class DataFrameSessionWindowingSuite extends QueryTest with SharedSparkSession
validateWindowColumnInSchema(schema2, "session")
}
}
+
+ test("SPARK-38349: No need to filter events when gapDuration greater than 0") {
+ // negative value
+ val df1 = Seq(
+ ("2016-03-27 19:39:30", 1, "a")).toDF("time", "value", "id")
+ .groupBy(session_window($"time", "-5 seconds"))
+ .agg(count("*").as("counts"))
+ .orderBy($"session_window.start".asc)
+ .select($"session_window.start".cast("string"), $"session_window.end".cast("string"),
+ $"counts")
+
+ val filter1 = df1.queryExecution.optimizedPlan.find(_.isInstanceOf[Filter])
+ assert(filter1.isDefined)
+ val exist1 = filter1.filter(_.toString.contains(">"))
+ assert(exist1.nonEmpty, "Need to filter windows when gapDuration less than 0")
+
+ // positive value
+ val df2 = Seq(
+ ("2016-03-27 19:39:40", 2, "a")).toDF("time", "value", "id")
+ .groupBy(session_window($"time", "5 seconds"))
+ .agg(count("*").as("counts"))
+ .orderBy($"session_window.start".asc)
+ .select($"session_window.start".cast("string"), $"session_window.end".cast("string"),
+ $"counts")
+
+ val filter2 = df2.queryExecution.optimizedPlan.find(_.isInstanceOf[Filter])
+ assert(filter2.isDefined)
+ val exist2 = filter2.filter(_.toString.contains(">"))
+ assert(exist2.isEmpty, "No need to filter windows when gapDuration value greater than 0")
+
+ // case when
+ val df3 = Seq(
+ ("2016-03-27 19:39:40", 2, "a")).toDF("time", "value", "id")
+ .groupBy(session_window($"time",
+ when(col("time").equalTo("1"), "5 seconds")
+ .when(col("time").equalTo("2"), "10 seconds")
+ .otherwise("10 seconds")))
+ .agg(count("*").as("counts"))
+ .orderBy($"session_window.start".asc)
+ .select($"session_window.start".cast("string"), $"session_window.end".cast("string"),
+ $"counts")
+
+ val filter3 = df3.queryExecution.optimizedPlan.find(_.isInstanceOf[Filter])
+ assert(filter3.isDefined)
+ val exist3 = filter3.filter(_.toString.contains(">"))
Review comment:
We shouldn't rely on string comparison to check query plan here.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] nyingping commented on pull request #35680: [SPARK-38349][SS] No need to filter events when sessionwindow gapDuration greater than 0
Posted by GitBox <gi...@apache.org>.
nyingping commented on pull request #35680:
URL: https://github.com/apache/spark/pull/35680#issuecomment-1079937369
I've narrowed it down to Static gap duration.
@HeartSaVioR @viirya Could you have a look when you have a chance?Thanks in advance!
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] viirya commented on a change in pull request #35680: [SPARK-38349][SS] No need to filter events when sessionwindow gapDuration greater than 0
Posted by GitBox <gi...@apache.org>.
viirya commented on a change in pull request #35680:
URL: https://github.com/apache/spark/pull/35680#discussion_r833828126
##########
File path: sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/Analyzer.scala
##########
@@ -4061,10 +4061,30 @@ object SessionWindowing extends Rule[LogicalPlan] {
case s: SessionWindow => sessionAttr
}
+ def isGapInvalid(child: Expression): Boolean = {
+ val calendarInterval = IntervalUtils
+ .safeStringToInterval(UTF8String.fromString(child.toString))
+ calendarInterval == null ||
+ calendarInterval.months + calendarInterval.days + calendarInterval.microseconds <= 0
+ }
+
+ val filterTimeSize = gapDuration match {
+ case gapDuration if gapDuration.find(_.isInstanceOf[ScalaUDF]).nonEmpty => true
+ case _ => gapDuration.child.children match {
+ case children if children.size > 0 =>
+ children.filter(_.dataType != BooleanType).exists(isGapInvalid(_))
+ case _ => isGapInvalid(gapDuration.child)
+ }
Review comment:
Seems we only need to check if the (casted) gap duration expression is foldable. If so, just evaluating it to get Calendar Interval value?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] HeartSaVioR commented on pull request #35680: [SPARK-38349][SS] No need to filter events when sessionwindow gapDuration greater than 0
Posted by GitBox <gi...@apache.org>.
HeartSaVioR commented on pull request #35680:
URL: https://github.com/apache/spark/pull/35680#issuecomment-1066313794
@nyingping
Now this PR has a small conflict; could you please resolve the conflict? Thanks in advance!
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] nyingping commented on a change in pull request #35680: [SPARK-38349][SS] No need to filter events when sessionwindow gapDuration greater than 0
Posted by GitBox <gi...@apache.org>.
nyingping commented on a change in pull request #35680:
URL: https://github.com/apache/spark/pull/35680#discussion_r833908883
##########
File path: sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/Analyzer.scala
##########
@@ -4061,10 +4061,30 @@ object SessionWindowing extends Rule[LogicalPlan] {
case s: SessionWindow => sessionAttr
}
+ def isGapInvalid(child: Expression): Boolean = {
+ val calendarInterval = IntervalUtils
+ .safeStringToInterval(UTF8String.fromString(child.toString))
Review comment:
@viirya
Thanks for reivew!
So far, I have judged two cases: static gap and `case when` expression. The case when expression filters the `BooleanType`. I'm not sure if there's anything missing.
Yes, I agree with your second point. I'm also trying to find a better way.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] HeartSaVioR commented on pull request #35680: [SPARK-38349][SS] No need to filter events when sessionwindow gapDuration greater than 0
Posted by GitBox <gi...@apache.org>.
HeartSaVioR commented on pull request #35680:
URL: https://github.com/apache/spark/pull/35680#issuecomment-1082569857
Thanks! Merging to master.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] HeartSaVioR closed pull request #35680: [SPARK-38349][SS] No need to filter events when sessionwindow gapDuration greater than 0
Posted by GitBox <gi...@apache.org>.
HeartSaVioR closed pull request #35680:
URL: https://github.com/apache/spark/pull/35680
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] HeartSaVioR commented on pull request #35680: [SPARK-38349][SS] No need to filter events when sessionwindow gapDuration greater than 0
Posted by GitBox <gi...@apache.org>.
HeartSaVioR commented on pull request #35680:
URL: https://github.com/apache/spark/pull/35680#issuecomment-1076317316
cc. @viirya since I see your client seems to use session window. For static gap duration this would reduce consistent overhead (filter with comparison between timestamp) per row.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] nyingping commented on a change in pull request #35680: [SPARK-38349][SS] No need to filter events when sessionwindow gapDuration greater than 0
Posted by GitBox <gi...@apache.org>.
nyingping commented on a change in pull request #35680:
URL: https://github.com/apache/spark/pull/35680#discussion_r833899929
##########
File path: sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/Analyzer.scala
##########
@@ -4061,10 +4061,30 @@ object SessionWindowing extends Rule[LogicalPlan] {
case s: SessionWindow => sessionAttr
}
+ def isGapInvalid(child: Expression): Boolean = {
+ val calendarInterval = IntervalUtils
+ .safeStringToInterval(UTF8String.fromString(child.toString))
+ calendarInterval == null ||
+ calendarInterval.months + calendarInterval.days + calendarInterval.microseconds <= 0
+ }
+
+ val filterTimeSize = gapDuration match {
Review comment:
@HeartSaVioR
Thanks for review!
I thought it is that there are three cases, including static gap duration, case when expression and user-defined function. When UDF, it needs to be filtered. Otherwise, there are gap duration and case when. In the case of case when, iterate its child nodes.I'm not quite sure if there are forms other than these three cases.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] viirya commented on a change in pull request #35680: [SPARK-38349][SS] No need to filter events when sessionwindow gapDuration greater than 0
Posted by GitBox <gi...@apache.org>.
viirya commented on a change in pull request #35680:
URL: https://github.com/apache/spark/pull/35680#discussion_r836809647
##########
File path: sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/Analyzer.scala
##########
@@ -4061,10 +4061,23 @@ object SessionWindowing extends Rule[LogicalPlan] {
case s: SessionWindow => sessionAttr
}
+ val filterTimeSize = gapDuration.child.dataType match {
Review comment:
yea, and we cannot simply cast it by `gapDuration.child.asInstanceOf[Literal]`. We should match if it is a literal.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] HyukjinKwon commented on pull request #35680: [SPARK-38349][SS] No need to filter events when sessionwindow gapDuration greater than 0
Posted by GitBox <gi...@apache.org>.
HyukjinKwon commented on pull request #35680:
URL: https://github.com/apache/spark/pull/35680#issuecomment-1056073252
cc @HeartSaVioR FYI
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] viirya commented on a change in pull request #35680: [SPARK-38349][SS] No need to filter events when sessionwindow gapDuration greater than 0
Posted by GitBox <gi...@apache.org>.
viirya commented on a change in pull request #35680:
URL: https://github.com/apache/spark/pull/35680#discussion_r833826333
##########
File path: sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/Analyzer.scala
##########
@@ -4061,10 +4061,30 @@ object SessionWindowing extends Rule[LogicalPlan] {
case s: SessionWindow => sessionAttr
}
+ def isGapInvalid(child: Expression): Boolean = {
+ val calendarInterval = IntervalUtils
+ .safeStringToInterval(UTF8String.fromString(child.toString))
Review comment:
`child.toString`? Does it work for all possible data types here?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] nyingping commented on a change in pull request #35680: [SPARK-38349][SS] No need to filter events when sessionwindow gapDuration greater than 0
Posted by GitBox <gi...@apache.org>.
nyingping commented on a change in pull request #35680:
URL: https://github.com/apache/spark/pull/35680#discussion_r833909866
##########
File path: sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/Analyzer.scala
##########
@@ -4061,10 +4061,30 @@ object SessionWindowing extends Rule[LogicalPlan] {
case s: SessionWindow => sessionAttr
}
+ def isGapInvalid(child: Expression): Boolean = {
+ val calendarInterval = IntervalUtils
+ .safeStringToInterval(UTF8String.fromString(child.toString))
+ calendarInterval == null ||
+ calendarInterval.months + calendarInterval.days + calendarInterval.microseconds <= 0
+ }
+
+ val filterTimeSize = gapDuration match {
+ case gapDuration if gapDuration.find(_.isInstanceOf[ScalaUDF]).nonEmpty => true
+ case _ => gapDuration.child.children match {
+ case children if children.size > 0 =>
+ children.filter(_.dataType != BooleanType).exists(isGapInvalid(_))
+ case _ => isGapInvalid(gapDuration.child)
+ }
Review comment:
Excluding `UDF` and `static gap`, it seems like this. Filter out `BooleanType`, which is calendar interval value.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] HeartSaVioR commented on a change in pull request #35680: [SPARK-38349][SS] No need to filter events when sessionwindow gapDuration greater than 0
Posted by GitBox <gi...@apache.org>.
HeartSaVioR commented on a change in pull request #35680:
URL: https://github.com/apache/spark/pull/35680#discussion_r836230538
##########
File path: sql/core/src/test/scala/org/apache/spark/sql/DataFrameSessionWindowingSuite.scala
##########
@@ -495,4 +495,65 @@ class DataFrameSessionWindowingSuite extends QueryTest with SharedSparkSession
validateWindowColumnInSchema(schema2, "session")
}
}
+
+ test("SPARK-38349: No need to filter events when gapDuration greater than 0") {
+ // negative value
Review comment:
Logic for verification can be deduplicated via adding inner-method in this test like below:
```
def checkFilterCondition(
logicalPlan: LogicalPlan,
expectTimeRange: Boolean,
assertHintMsg: String): Unit = {
val filter = logicalPlan.find { plan =>
plan.isInstanceOf[Filter] && plan.children.head.isInstanceOf[Project]
}
assert(filter.isDefined)
val exist = filter.get.expressions.flatMap { expr =>
expr.collect { case gt: GreaterThan => gt }
}
if (expectTimeRange) {
assert(exist.nonEmpty, assertHintMsg)
} else {
assert(exist.isEmpty, assertHintMsg)
}
}
```
and call the method per case.
Let's use analyzed plan (`df.queryExecution.logical`) instead of optimizedPlan which Spark can optimize the plan and may have changed the plan out of the expectation.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] nyingping commented on pull request #35680: [SPARK-38349][SS] No need to filter events when sessionwindow gapDuration greater than 0
Posted by GitBox <gi...@apache.org>.
nyingping commented on pull request #35680:
URL: https://github.com/apache/spark/pull/35680#issuecomment-1081509735
@viirya @HeartSaVioR
Thanks for suggestion,Thanks a lot!
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] HeartSaVioR commented on a change in pull request #35680: [SPARK-38349][SS] No need to filter events when sessionwindow gapDuration greater than 0
Posted by GitBox <gi...@apache.org>.
HeartSaVioR commented on a change in pull request #35680:
URL: https://github.com/apache/spark/pull/35680#discussion_r837181916
##########
File path: sql/core/src/test/scala/org/apache/spark/sql/DataFrameSessionWindowingSuite.scala
##########
@@ -495,4 +495,56 @@ class DataFrameSessionWindowingSuite extends QueryTest with SharedSparkSession
validateWindowColumnInSchema(schema2, "session")
}
}
+
+ test("SPARK-38349: No need to filter events when gapDuration greater than 0") {
+ // negative gap duration
+ check("-5 seconds", true, "Need to filter events when gap duration less than 0")
+
+ // positive gap duration
+ check("5 seconds", false, "No need to filter events when gap duration greater than 0")
+
+ // invalid gap duration
+ check("x seconds", true, "Need to filter events when gap duration invalid")
+
+ // dynamic gap duration
+ check(when(col("time").equalTo("1"), "5 seconds")
+ .when(col("time").equalTo("2"), "10 seconds")
+ .otherwise("10 seconds"), true, "Need to filter events when gap duration dynamically")
+
+ def check(
+ gapDuration: Any,
+ expectTimeRange: Boolean,
+ assertHintMsg: String): Unit = {
+ val data = Seq(
+ ("2016-03-27 19:39:30", 1, "a")).toDF("time", "value", "id")
+ val df = if (gapDuration.isInstanceOf[String]) {
Review comment:
seems OK since this is a simple test code
##########
File path: sql/core/src/test/scala/org/apache/spark/sql/DataFrameSessionWindowingSuite.scala
##########
@@ -495,4 +495,56 @@ class DataFrameSessionWindowingSuite extends QueryTest with SharedSparkSession
validateWindowColumnInSchema(schema2, "session")
}
}
+
+ test("SPARK-38349: No need to filter events when gapDuration greater than 0") {
+ // negative gap duration
+ check("-5 seconds", true, "Need to filter events when gap duration less than 0")
+
+ // positive gap duration
+ check("5 seconds", false, "No need to filter events when gap duration greater than 0")
+
+ // invalid gap duration
+ check("x seconds", true, "Need to filter events when gap duration invalid")
+
+ // dynamic gap duration
+ check(when(col("time").equalTo("1"), "5 seconds")
+ .when(col("time").equalTo("2"), "10 seconds")
+ .otherwise("10 seconds"), true, "Need to filter events when gap duration dynamically")
+
+ def check(
+ gapDuration: Any,
+ expectTimeRange: Boolean,
+ assertHintMsg: String): Unit = {
+ val data = Seq(
+ ("2016-03-27 19:39:30", 1, "a")).toDF("time", "value", "id")
+ val df = if (gapDuration.isInstanceOf[String]) {
+ data.groupBy(session_window($"time", gapDuration.asInstanceOf[String]))
+ } else {
+ data.groupBy(session_window($"time", gapDuration.asInstanceOf[Column]))
+ }
+ val aggregate = df.agg(count("*").as("counts"))
+ .select($"session_window.start".cast("string"), $"session_window.end".cast("string"),
+ $"counts")
+
+ checkFilterCondition(aggregate.queryExecution.logical, expectTimeRange, assertHintMsg)
+ }
+
+ def checkFilterCondition(
+ logicalPlan: LogicalPlan,
Review comment:
nit: indentation is off (4 spaces)
##########
File path: sql/core/src/test/scala/org/apache/spark/sql/DataFrameSessionWindowingSuite.scala
##########
@@ -495,4 +495,56 @@ class DataFrameSessionWindowingSuite extends QueryTest with SharedSparkSession
validateWindowColumnInSchema(schema2, "session")
}
}
+
+ test("SPARK-38349: No need to filter events when gapDuration greater than 0") {
+ // negative gap duration
+ check("-5 seconds", true, "Need to filter events when gap duration less than 0")
+
+ // positive gap duration
+ check("5 seconds", false, "No need to filter events when gap duration greater than 0")
+
+ // invalid gap duration
+ check("x seconds", true, "Need to filter events when gap duration invalid")
+
+ // dynamic gap duration
+ check(when(col("time").equalTo("1"), "5 seconds")
+ .when(col("time").equalTo("2"), "10 seconds")
+ .otherwise("10 seconds"), true, "Need to filter events when gap duration dynamically")
+
+ def check(
+ gapDuration: Any,
Review comment:
nit: indentation is off (4 spaces)
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] viirya commented on a change in pull request #35680: [SPARK-38349][SS] No need to filter events when sessionwindow gapDuration greater than 0
Posted by GitBox <gi...@apache.org>.
viirya commented on a change in pull request #35680:
URL: https://github.com/apache/spark/pull/35680#discussion_r836811142
##########
File path: sql/core/src/test/scala/org/apache/spark/sql/DataFrameSessionWindowingSuite.scala
##########
@@ -495,4 +495,65 @@ class DataFrameSessionWindowingSuite extends QueryTest with SharedSparkSession
validateWindowColumnInSchema(schema2, "session")
}
}
+
+ test("SPARK-38349: No need to filter events when gapDuration greater than 0") {
+ // negative value
+ val df1 = Seq(
+ ("2016-03-27 19:39:30", 1, "a")).toDF("time", "value", "id")
+ .groupBy(session_window($"time", "-5 seconds"))
+ .agg(count("*").as("counts"))
+ .orderBy($"session_window.start".asc)
+ .select($"session_window.start".cast("string"), $"session_window.end".cast("string"),
+ $"counts")
+
+ val filter1 = df1.queryExecution.optimizedPlan.find(_.isInstanceOf[Filter])
+ assert(filter1.isDefined)
+ val exist1 = filter1.filter(_.toString.contains(">"))
+ assert(exist1.nonEmpty, "Need to filter windows when gapDuration less than 0")
+
+ // positive value
+ val df2 = Seq(
+ ("2016-03-27 19:39:40", 2, "a")).toDF("time", "value", "id")
+ .groupBy(session_window($"time", "5 seconds"))
+ .agg(count("*").as("counts"))
+ .orderBy($"session_window.start".asc)
+ .select($"session_window.start".cast("string"), $"session_window.end".cast("string"),
+ $"counts")
+
+ val filter2 = df2.queryExecution.optimizedPlan.find(_.isInstanceOf[Filter])
+ assert(filter2.isDefined)
+ val exist2 = filter2.filter(_.toString.contains(">"))
+ assert(exist2.isEmpty, "No need to filter windows when gapDuration value greater than 0")
+
+ // case when
+ val df3 = Seq(
+ ("2016-03-27 19:39:40", 2, "a")).toDF("time", "value", "id")
+ .groupBy(session_window($"time",
+ when(col("time").equalTo("1"), "5 seconds")
+ .when(col("time").equalTo("2"), "10 seconds")
+ .otherwise("10 seconds")))
+ .agg(count("*").as("counts"))
+ .orderBy($"session_window.start".asc)
+ .select($"session_window.start".cast("string"), $"session_window.end".cast("string"),
+ $"counts")
+
+ val filter3 = df3.queryExecution.optimizedPlan.find(_.isInstanceOf[Filter])
+ assert(filter3.isDefined)
+ val exist3 = filter3.filter(_.toString.contains(">"))
Review comment:
Please follow @HeartSaVioR's suggestion to check condition expression in Filter.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] nyingping commented on pull request #35680: [SPARK-38349][SS] No need to filter events when sessionwindow gapDuration greater than 0
Posted by GitBox <gi...@apache.org>.
nyingping commented on pull request #35680:
URL: https://github.com/apache/spark/pull/35680#issuecomment-1066318133
@HeartSaVioR yeah,resolved.thanks for your time!
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] HeartSaVioR commented on a change in pull request #35680: [SPARK-38349][SS] No need to filter events when sessionwindow gapDuration greater than 0
Posted by GitBox <gi...@apache.org>.
HeartSaVioR commented on a change in pull request #35680:
URL: https://github.com/apache/spark/pull/35680#discussion_r833192788
##########
File path: sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/Analyzer.scala
##########
@@ -4061,10 +4061,30 @@ object SessionWindowing extends Rule[LogicalPlan] {
case s: SessionWindow => sessionAttr
}
+ def isGapInvalid(child: Expression): Boolean = {
+ val calendarInterval = IntervalUtils
+ .safeStringToInterval(UTF8String.fromString(child.toString))
+ calendarInterval == null ||
+ calendarInterval.months + calendarInterval.days + calendarInterval.microseconds <= 0
+ }
+
+ val filterTimeSize = gapDuration match {
Review comment:
Which cases this targets to? I only see the case of StringType (meaning static gap) which is going to be casted as CalendarIntervalType, but please correct me if I'm missing something.
Please make sure the code tightly picks up with the cases. For me, current pattern matching does not seem to be intuitive to understand. If you feel this code does not seem to be intuitive due to the fact you are disassembling Cast here, you can just deal with `session.gapDuration` instead, which is not yet wrapped.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] HeartSaVioR commented on pull request #35680: [SPARK-38349][SS] No need to filter events when sessionwindow gapDuration greater than 0
Posted by GitBox <gi...@apache.org>.
HeartSaVioR commented on pull request #35680:
URL: https://github.com/apache/spark/pull/35680#issuecomment-1077299616
I guess this is being complicated to also handle the case of dynamic gap with "case when" & literals.
What about just narrowing down the scope to the static gap? This is a majority of use case for session window, and this will simplify the logic a lot - probably need to handle Literal only.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org