[SPARK-40925][SQL][SS] Fix stateful operator late record filtering #38405

alex-balikov · 2022-10-26T22:19:01Z

What changes were proposed in this pull request?

This PR fixes the input late record filtering done by stateful operators to allow for chaining of stateful operators. Currently stateful operators are initialized with the current microbatch watermark and perform both input late record filtering and state eviction (e.g. producing aggregations) using the same watermark value. The state evicted (or aggregates produced) due to watermark advancing is behind the watermark and thus effectively late - if a following stateful operator consumes the output of the previous one, the input records will be filtered as late.

This PR provides two watermark values to the stateful operators - one from the previous microbatch to be used for late record filtering and the one from the current microbatch (as in the existing code) to be used for state eviction. This solves the above problem of the broken late record filtering.

Note that this PR still does not solve the issue of time-interval stream join producing records delayed against the watermark. Therefore time-interval streaming join followed by stateful operators is still not supported. That will be fixed in a follow up PR (and a SPIP) effectively replacing the single global watermark with conceptually watermarks per operator.

Also, the stateful operator chains unblocked by this PR (e.g. a chain of window aggregations) are still blocked by the unsupported operations checker. The new test for these scenarios - MultiStatefulOperatorsSuite has to explicitly disable the unsupported ops check. This again will be fixed in a follow-up PR.

Why are the changes needed?

The PR allows Spark Structured Streaming to support chaining of stateful operators e.g. chaining of time window aggregations which is a meaningful streaming scenario.

Does this PR introduce any user-facing change?

With this PR, chains of stateful operators will be supported in Spark Structured Streaming.

How was this patch tested?

Added a new test suite - MultiStatefulOperatorsSuite

…-ops-base

HeartSaVioR

Looks OK in overall. Left comments.

sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/OffsetSeqLog.scala

sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/statefulOperators.scala

sql/core/src/test/scala/org/apache/spark/sql/streaming/MultiStatefulOperatorsSuite.scala

sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/statefulOperators.scala

sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/sources/MicroBatchWrite.scala

HeartSaVioR · 2022-10-27T08:18:35Z

sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/SessionWindow.scala

@@ -68,11 +68,29 @@ case class SessionWindow(timeColumn: Expression, gapDuration: Expression) extend
  with Unevaluable
  with NonSQLExpression {

+  private def inputTypeOnTimeColumn: AbstractDataType = {


(Oh I missed that the change from rule side has already merged as a part of introduction of window_time.)

Sorry, are you asking for anything actionable?

Nope. Just for future visibility.

…ng/OffsetSeqLog.scala Co-authored-by: Jungtaek Lim <kabhwan.opensource@gmail.com>

…-ops-base

…balikov/spark into multiple_stateful-ops-base

AmplabJenkins · 2022-10-27T19:56:59Z

Can one of the admins verify this patch?

…-ops-base

HeartSaVioR

+1
I'll leave this PR in a couple of days to see whether we can get another eyes of review. I'll merge this in early next week there is no outstanding comment.

HeartSaVioR · 2022-10-31T01:12:28Z

OK, no outstanding comment so far. Thanks! Merging to master.

…ueries ### What changes were proposed in this pull request? As a followup to [SPARK-40925], [github PR](#38405), Remove corresponding checks in UnsupportedOperationChecker so that customers don't have to explicitly add new conf withSQLConf("spark.sql.streaming.unsupportedOperationCheck" -> "false") to use the new multi-stateful operators. In other words we are enabling multi-stateful operators by default. As a side effect, the API of `checkStreamingQueryGlobalWatermarkLimit(LogicalPlan, OutputMode)` is also changed to `checkStreamingQueryGlobalWatermarkLimit(LogicalPlan)` New tests are added to `MultiStatefulOperatorsSuite.scala`, but I could also add equivalent ones to `UnsupportedOperationsSuite.scala` if requested. ### Why are the changes needed? To enable new multiple-stateful operators by default. Right now users need to set SQL conf `unsupportedOperationCheck` to false explicitly, which also disables many other useful checks. ### Does this PR introduce _any_ user-facing change? No. All current running queries won't be impacted. But new queries could use chained stateful operators. ### How was this patch tested? Unit Tests. Closes #38503 from WweiL/SPARK-40940-multi-state-checkers. Authored-by: Wei Liu <wei.liu@databricks.com> Signed-off-by: Jungtaek Lim <kabhwan.opensource@gmail.com>

### What changes were proposed in this pull request? This PR fixes the input late record filtering done by stateful operators to allow for chaining of stateful operators. Currently stateful operators are initialized with the current microbatch watermark and perform both input late record filtering and state eviction (e.g. producing aggregations) using the same watermark value. The state evicted (or aggregates produced) due to watermark advancing is behind the watermark and thus effectively late - if a following stateful operator consumes the output of the previous one, the input records will be filtered as late. This PR provides two watermark values to the stateful operators - one from the previous microbatch to be used for late record filtering and the one from the current microbatch (as in the existing code) to be used for state eviction. This solves the above problem of the broken late record filtering. Note that this PR still does not solve the issue of time-interval stream join producing records delayed against the watermark. Therefore time-interval streaming join followed by stateful operators is still not supported. That will be fixed in a follow up PR (and a SPIP) effectively replacing the single global watermark with conceptually watermarks per operator. Also, the stateful operator chains unblocked by this PR (e.g. a chain of window aggregations) are still blocked by the unsupported operations checker. The new test for these scenarios - MultiStatefulOperatorsSuite has to explicitly disable the unsupported ops check. This again will be fixed in a follow-up PR. ### Why are the changes needed? The PR allows Spark Structured Streaming to support chaining of stateful operators e.g. chaining of time window aggregations which is a meaningful streaming scenario. ### Does this PR introduce _any_ user-facing change? With this PR, chains of stateful operators will be supported in Spark Structured Streaming. ### How was this patch tested? Added a new test suite - MultiStatefulOperatorsSuite Closes apache#38405 from alex-balikov/multiple_stateful-ops-base. Lead-authored-by: Alex Balikov <91913242+alex-balikov@users.noreply.github.com> Co-authored-by: Alex Balikov <alex.balikov@databricks.com> Signed-off-by: Jungtaek Lim <kabhwan.opensource@gmail.com>

…ueries ### What changes were proposed in this pull request? As a followup to [SPARK-40925], [github PR](apache#38405), Remove corresponding checks in UnsupportedOperationChecker so that customers don't have to explicitly add new conf withSQLConf("spark.sql.streaming.unsupportedOperationCheck" -> "false") to use the new multi-stateful operators. In other words we are enabling multi-stateful operators by default. As a side effect, the API of `checkStreamingQueryGlobalWatermarkLimit(LogicalPlan, OutputMode)` is also changed to `checkStreamingQueryGlobalWatermarkLimit(LogicalPlan)` New tests are added to `MultiStatefulOperatorsSuite.scala`, but I could also add equivalent ones to `UnsupportedOperationsSuite.scala` if requested. ### Why are the changes needed? To enable new multiple-stateful operators by default. Right now users need to set SQL conf `unsupportedOperationCheck` to false explicitly, which also disables many other useful checks. ### Does this PR introduce _any_ user-facing change? No. All current running queries won't be impacted. But new queries could use chained stateful operators. ### How was this patch tested? Unit Tests. Closes apache#38503 from WweiL/SPARK-40940-multi-state-checkers. Authored-by: Wei Liu <wei.liu@databricks.com> Signed-off-by: Jungtaek Lim <kabhwan.opensource@gmail.com>

alex-balikov and others added 10 commits October 13, 2022 14:20

x

acc76dc

Merge remote-tracking branch 'upstream/master' into multiple_stateful…

8831691

…-ops-base

f

59002a2

Merge remote-tracking branch 'upstream/master' into multiple_stateful…

3f1c322

…-ops-base

Merge remote-tracking branch 'upstream/master' into multiple_stateful…

179f422

…-ops-base

build fixes

6262312

Merge remote-tracking branch 'upstream/master' into multiple_stateful…

7c1f066

…-ops-base

revert changes

d031e32

Merge remote-tracking branch 'upstream/master' into multiple_stateful…

c789d19

…-ops-base

fixes

4e12828

github-actions bot added PYTHON SQL STRUCTURED STREAMING labels Oct 26, 2022

fixes

36b6826

mridulm changed the title ~~[SPARK-40925][SQL][CORE][SS]Fix stateful operator late record filtering~~ [SPARK-40925][SQL][SS]Fix stateful operator late record filtering Oct 27, 2022

HeartSaVioR changed the title ~~[SPARK-40925][SQL][SS]Fix stateful operator late record filtering~~ [SPARK-40925][SQL][SS] Fix stateful operator late record filtering Oct 27, 2022

HeartSaVioR reviewed Oct 27, 2022

View reviewed changes

alex-balikov and others added 4 commits October 27, 2022 11:35

Update sql/core/src/main/scala/org/apache/spark/sql/execution/streami…

764be57

…ng/OffsetSeqLog.scala Co-authored-by: Jungtaek Lim <kabhwan.opensource@gmail.com>

Merge remote-tracking branch 'upstream/master' into multiple_stateful…

5947fc9

…-ops-base

code review fixes

bee6185

Merge branch 'multiple_stateful-ops-base' of https://github.com/alex-…

0e96fae

…balikov/spark into multiple_stateful-ops-base

alex-balikov added 5 commits October 27, 2022 13:15

code review fixes

565f8af

fixes

24f1b61

fixes

9515f10

fixes

03cbdd6

fixes

572263d

alex-balikov requested a review from HeartSaVioR October 27, 2022 23:39

Merge remote-tracking branch 'upstream/master' into multiple_stateful…

68b14e3

…-ops-base

HeartSaVioR approved these changes Oct 28, 2022

View reviewed changes

fixes

6a023ae

HeartSaVioR closed this in 406d0e2 Oct 31, 2022

WweiL mentioned this pull request Nov 3, 2022

[SPARK-40940] Remove Multi-stateful operator checkers for streaming queries. #38503

Closed

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

[SPARK-40925][SQL][SS] Fix stateful operator late record filtering #38405

[SPARK-40925][SQL][SS] Fix stateful operator late record filtering #38405

alex-balikov commented Oct 26, 2022

HeartSaVioR left a comment

HeartSaVioR Oct 27, 2022 •

edited

Loading

alex-balikov Oct 28, 2022

HeartSaVioR Oct 28, 2022

AmplabJenkins commented Oct 27, 2022

HeartSaVioR left a comment

HeartSaVioR commented Oct 31, 2022

[SPARK-40925][SQL][SS] Fix stateful operator late record filtering #38405

[SPARK-40925][SQL][SS] Fix stateful operator late record filtering #38405

Conversation

alex-balikov commented Oct 26, 2022

What changes were proposed in this pull request?

Why are the changes needed?

Does this PR introduce any user-facing change?

How was this patch tested?

HeartSaVioR left a comment

Choose a reason for hiding this comment

HeartSaVioR Oct 27, 2022 • edited Loading

Choose a reason for hiding this comment

alex-balikov Oct 28, 2022

Choose a reason for hiding this comment

HeartSaVioR Oct 28, 2022

Choose a reason for hiding this comment

AmplabJenkins commented Oct 27, 2022

HeartSaVioR left a comment

Choose a reason for hiding this comment

HeartSaVioR commented Oct 31, 2022

HeartSaVioR Oct 27, 2022 •

edited

Loading