[SPARK-34892][SS] Introduce MergingSortWithSessionWindowStateIterator sorting input rows and rows in state efficiently #33077

HeartSaVioR · 2021-06-25T05:48:54Z

Introduction: this PR is a part of SPARK-10816 (EventTime based sessionization (session window)). Please refer #31937 to see the overall view of the code change. (Note that code diff could be diverged a bit.)

What changes were proposed in this pull request?

This PR introduces MergingSortWithSessionWindowStateIterator, which does "merge sort" between input rows and sessions in state based on group key and session's start time.

Note that the iterator does merge sort among input rows and sessions grouped by grouping key. The iterator doesn't provide sessions in state which keys don't exist in input rows. For input rows, the iterator will provide all rows regardless of the existence of matching sessions in state.

MergingSortWithSessionWindowStateIterator works on the precondition that given iterator is sorted by "group keys + start time of session window", and the iterator still retains the characteristic of the sort.

Why are the changes needed?

This part is a one of required on implementing SPARK-10816.

Does this PR introduce any user-facing change?

No.

How was this patch tested?

New UT added.

HeartSaVioR · 2021-06-25T05:50:11Z

Same here; I marked this as draft as other PRs has to be reviewed and merged earlier. I'll rebase this PR once all other PRs are merged.

SparkQA · 2021-06-25T06:40:20Z

Kubernetes integration test starting
URL: https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder-K8s/44837/

SparkQA · 2021-06-25T07:14:47Z

Kubernetes integration test status success
URL: https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder-K8s/44837/

SparkQA · 2021-06-25T08:42:04Z

Kubernetes integration test starting
URL: https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder-K8s/44843/

SparkQA · 2021-06-25T09:17:25Z

Kubernetes integration test status success
URL: https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder-K8s/44843/

SparkQA · 2021-06-25T11:05:42Z

Test build #140306 has finished for PR 33077 at commit eb433d7.

This patch passes all tests.
This patch merges cleanly.
This patch adds the following public classes (experimental):
class MergingSortWithSessionWindowStateIterator(

SparkQA · 2021-06-25T12:16:17Z

Test build #140311 has finished for PR 33077 at commit 83cc78a.

This patch passes all tests.
This patch merges cleanly.
This patch adds the following public classes (experimental):
class MergingSortWithSessionWindowStateIterator(

HeartSaVioR · 2021-07-13T20:08:31Z

cc. @viirya @xuanyuanking Please take a look. Thanks!

viirya · 2021-07-13T20:09:49Z

Thanks @HeartSaVioR. I will review this today.

SparkQA · 2021-07-13T20:39:03Z

Test build #140981 has finished for PR 33077 at commit 60f6114.

This patch fails to build.
This patch merges cleanly.
This patch adds no public classes.

SparkQA · 2021-07-13T20:47:06Z

Kubernetes integration test unable to build dist.

exiting with code: 1
URL: https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder-K8s/45495/

SparkQA · 2021-07-14T00:27:00Z

Kubernetes integration test starting
URL: https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder-K8s/45501/

SparkQA · 2021-07-14T01:01:28Z

Kubernetes integration test status success
URL: https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder-K8s/45501/

… sorting input rows and rows in state efficiently

SparkQA · 2021-07-14T02:08:40Z

Kubernetes integration test unable to build dist.

exiting with code: 1
URL: https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder-K8s/45503/

SparkQA · 2021-07-14T03:51:24Z

Test build #140987 has finished for PR 33077 at commit b540632.

This patch passes all tests.
This patch merges cleanly.
This patch adds no public classes.

SparkQA · 2021-07-14T05:43:31Z

Test build #140989 has finished for PR 33077 at commit e4a74a3.

This patch passes all tests.
This patch merges cleanly.
This patch adds the following public classes (experimental):
class MergingSortWithSessionWindowStateIterator(

viirya · 2021-07-14T06:05:36Z

...ala/org/apache/spark/sql/execution/streaming/MergingSortWithSessionWindowStateIterator.scala

+  private var currentRow: SessionRowInformation = _
+  private var currentStateRow: SessionRowInformation = _
+  private var currentStateIter: Iterator[InternalRow] = _
+  private var currentStateFetchedKey: UnsafeRow = _


A few suggestions:

currentRow -> currentRowFromInput
currentStateRow -> currentRowFromState
currentStateIter -> sessionIterFromState
currentStateFetchedKey -> currentSessionKey

And maybe add a few comments explaining these variables for readability.

viirya · 2021-07-14T06:10:03Z

...ala/org/apache/spark/sql/execution/streaming/MergingSortWithSessionWindowStateIterator.scala

+      } else {
+        // compare
+        if (currentRow.keys != currentStateRow.keys) {
+          // state row cannot advance to row in input, so state row should be lower


Does this case mean, the input iterator advances to new keys other than current sessions from the state? So we should output from current sessions until it ends and retrieves new sessions from the state again?

Exactly. We retrieve state rows for the specific key only when there's a new key from input side, so the case is not possible state side advances compared to input side. If the keys differ, there're rows to process in state side. The opposite case is not possible.

viirya · 2021-07-14T06:14:02Z

...rg/apache/spark/sql/execution/streaming/MergingSortWithSessionWindowStateIteratorSuite.scala

+        rowAttributes)
+
+      val actual = iter.map(_.copy()).toList
+      assert(actual.isEmpty)


Hm? Why it is empty? If input is empty, doesn't it output sorted sessions in the state?

Oh, nvm. I see.

viirya

Looks good. I have a few suggestions about variable names.

viirya · 2021-07-14T07:34:25Z

I will merge this tomorrow. Thanks @HeartSaVioR

SparkQA · 2021-07-14T08:22:44Z

Kubernetes integration test starting
URL: https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder-K8s/45520/

SparkQA · 2021-07-14T08:56:52Z

Kubernetes integration test status success
URL: https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder-K8s/45520/

HeartSaVioR · 2021-07-14T09:46:54Z

Please bear with me on merging this one now to unblock the last PR #33081. We can do post-review for any part of session window changes even it's merged during QA period (and even RC period if someone catches defect.)

… sorting input rows and rows in state efficiently Introduction: this PR is a part of SPARK-10816 (EventTime based sessionization (session window)). Please refer #31937 to see the overall view of the code change. (Note that code diff could be diverged a bit.) ### What changes were proposed in this pull request? This PR introduces MergingSortWithSessionWindowStateIterator, which does "merge sort" between input rows and sessions in state based on group key and session's start time. Note that the iterator does merge sort among input rows and sessions grouped by grouping key. The iterator doesn't provide sessions in state which keys don't exist in input rows. For input rows, the iterator will provide all rows regardless of the existence of matching sessions in state. MergingSortWithSessionWindowStateIterator works on the precondition that given iterator is sorted by "group keys + start time of session window", and the iterator still retains the characteristic of the sort. ### Why are the changes needed? This part is a one of required on implementing SPARK-10816. ### Does this PR introduce _any_ user-facing change? No. ### How was this patch tested? New UT added. Closes #33077 from HeartSaVioR/SPARK-34892-SPARK-10816-PR-31570-part-4. Authored-by: Jungtaek Lim <[email protected]> Signed-off-by: Jungtaek Lim <[email protected]> (cherry picked from commit 12a576f) Signed-off-by: Jungtaek Lim <[email protected]>

HeartSaVioR · 2021-07-14T09:49:41Z

Thanks @viirya for reviewing! I merged this to master/3.2. I'll rebase #33081 to reflect the latest master branch.

SparkQA · 2021-07-14T11:47:21Z

Test build #141006 has finished for PR 33077 at commit c674046.

This patch passes all tests.
This patch merges cleanly.
This patch adds no public classes.

… sorting input rows and rows in state efficiently Introduction: this PR is a part of SPARK-10816 (EventTime based sessionization (session window)). Please refer apache#31937 to see the overall view of the code change. (Note that code diff could be diverged a bit.) ### What changes were proposed in this pull request? This PR introduces MergingSortWithSessionWindowStateIterator, which does "merge sort" between input rows and sessions in state based on group key and session's start time. Note that the iterator does merge sort among input rows and sessions grouped by grouping key. The iterator doesn't provide sessions in state which keys don't exist in input rows. For input rows, the iterator will provide all rows regardless of the existence of matching sessions in state. MergingSortWithSessionWindowStateIterator works on the precondition that given iterator is sorted by "group keys + start time of session window", and the iterator still retains the characteristic of the sort. ### Why are the changes needed? This part is a one of required on implementing SPARK-10816. ### Does this PR introduce _any_ user-facing change? No. ### How was this patch tested? New UT added. Closes apache#33077 from HeartSaVioR/SPARK-34892-SPARK-10816-PR-31570-part-4. Authored-by: Jungtaek Lim <[email protected]> Signed-off-by: Jungtaek Lim <[email protected]>

HeartSaVioR marked this pull request as draft June 25, 2021 05:49

github-actions bot added SQL STRUCTURED STREAMING labels Jun 25, 2021

HeartSaVioR force-pushed the SPARK-34892-SPARK-10816-PR-31570-part-4 branch from eb433d7 to 83cc78a Compare June 25, 2021 07:43

HeartSaVioR force-pushed the SPARK-34892-SPARK-10816-PR-31570-part-4 branch from 83cc78a to abab1e8 Compare July 13, 2021 19:54

HeartSaVioR marked this pull request as ready for review July 13, 2021 19:54

[SPARK-34892][SS] Introduce MergingSortWithSessionWindowStateIterator…

e4a74a3

… sorting input rows and rows in state efficiently

HeartSaVioR force-pushed the SPARK-34892-SPARK-10816-PR-31570-part-4 branch from b540632 to e4a74a3 Compare July 14, 2021 01:22

viirya reviewed Jul 14, 2021

View reviewed changes

Reflect review comment

c674046

viirya approved these changes Jul 14, 2021

View reviewed changes

HeartSaVioR closed this in 12a576f Jul 14, 2021

[SPARK-34892][SS] Introduce MergingSortWithSessionWindowStateIterator sorting input rows and rows in state efficiently #33077

[SPARK-34892][SS] Introduce MergingSortWithSessionWindowStateIterator sorting input rows and rows in state efficiently #33077

Uh oh!

Conversation

HeartSaVioR commented Jun 25, 2021

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?

Uh oh!

HeartSaVioR commented Jun 25, 2021

Uh oh!

SparkQA commented Jun 25, 2021

Uh oh!

SparkQA commented Jun 25, 2021

Uh oh!

SparkQA commented Jun 25, 2021

Uh oh!

SparkQA commented Jun 25, 2021

Uh oh!

SparkQA commented Jun 25, 2021

Uh oh!

SparkQA commented Jun 25, 2021

Uh oh!

HeartSaVioR commented Jul 13, 2021

Uh oh!

viirya commented Jul 13, 2021

Uh oh!

SparkQA commented Jul 13, 2021

Uh oh!

SparkQA commented Jul 13, 2021

Uh oh!

SparkQA commented Jul 14, 2021

Uh oh!

SparkQA commented Jul 14, 2021

Uh oh!

SparkQA commented Jul 14, 2021

Uh oh!

SparkQA commented Jul 14, 2021

Uh oh!

SparkQA commented Jul 14, 2021

Uh oh!

viirya Jul 14, 2021

Choose a reason for hiding this comment

Uh oh!

viirya Jul 14, 2021

Choose a reason for hiding this comment

Uh oh!

viirya Jul 14, 2021

Choose a reason for hiding this comment

Uh oh!

HeartSaVioR Jul 14, 2021 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

viirya Jul 14, 2021

Choose a reason for hiding this comment

Uh oh!

viirya Jul 14, 2021

Choose a reason for hiding this comment

Uh oh!

viirya left a comment

Choose a reason for hiding this comment

Uh oh!

viirya commented Jul 14, 2021

Uh oh!

SparkQA commented Jul 14, 2021

Uh oh!

SparkQA commented Jul 14, 2021

Uh oh!

HeartSaVioR commented Jul 14, 2021

Uh oh!

HeartSaVioR commented Jul 14, 2021 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Uh oh!

SparkQA commented Jul 14, 2021

Uh oh!

Reviewers

Assignees

Labels

HeartSaVioR Jul 14, 2021 •

edited

Loading

HeartSaVioR commented Jul 14, 2021 •

edited

Loading