[SPARK-34893][SS] Support session window natively #33081

HeartSaVioR · 2021-06-25T08:29:51Z

Introduction: this PR is the last 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 proposes to support native session window. Please refer the comments/design doc in SPARK-10816 for more details on the rationalization and design (could be outdated a bit compared to the PR).

The definition of the boundary of "session window" is [the timestamp of start event ~ the timestamp of last event + gap duration). That said, unlike time window, session window is a dynamic window which can expand if new input row is added to the session. To handle expansion of session window, Spark defines session window per input row, and "merge" windows if they can be merged (boundaries are overlapped).

This PR leverages two different approaches on merging session windows:

merging session windows with Spark's aggregation logic (a variant of sort aggregation)
updating session window for all rows bound to the same session, and applying aggregation logic afterwards

First one is preferable as it outperforms compared to the second one, though it can be only used if merging session window can be applied altogether with aggregation. It is not applicable on all the cases, so second one is used to cover the remaining cases.

This PR also applies the optimization on merging input rows and existing sessions with retaining the order (group keys + start timestamp of session window), leveraging the fact the number of existing sessions per group key won't be huge.

The state format is versioned, so that we can bring a new state format if we find a better one.

Why are the changes needed?

For now, to deal with sessionization, Spark requires end users to play with (flat)MapGroupsWithState directly which has a couple of major drawbacks:

(flat)MapGroupsWithState is lower level API and end users have to code everything in details for defining session window and merging windows
built-in aggregate functions cannot be used and end users have to deal with aggregation by themselves
(flat)MapGroupsWithState is only available in Scala/Java.

With native support of session window, end users simply use "session_window" like they use "window" for tumbling/sliding window, and leverage built-in aggregate functions as well as UDAFs to simply define aggregations.

Quoting the query example from test suite:

    val inputData = MemoryStream[(String, Long)]

    // Split the lines into words, treat words as sessionId of events
    val events = inputData.toDF()
      .select($"_1".as("value"), $"_2".as("timestamp"))
      .withColumn("eventTime", $"timestamp".cast("timestamp"))
      .selectExpr("explode(split(value, ' ')) AS sessionId", "eventTime")
      .withWatermark("eventTime", "30 seconds")

    val sessionUpdates = events
      .groupBy(session_window($"eventTime", "10 seconds") as 'session, 'sessionId)
      .agg(count("*").as("numEvents"))
      .selectExpr("sessionId", "CAST(session.start AS LONG)", "CAST(session.end AS LONG)",
        "CAST(session.end AS LONG) - CAST(session.start AS LONG) AS durationMs",
        "numEvents")

which is same as StructuredSessionization (native session window is shorter and clearer even ignoring model classes).

spark/examples/src/main/scala/org/apache/spark/examples/sql/streaming/StructuredSessionization.scala

Lines 66 to 105 in 39542bb

    
           // Split the lines into words, treat words as sessionId of events 
        
           val events = lines 
        
             .as[(String, Timestamp)] 
        
             .flatMap { case (line, timestamp) => 
        
               line.split(" ").map(word => Event(sessionId = word, timestamp)) 
        
             } 
        
           // Sessionize the events. Track number of events, start and end timestamps of session, 
        
           // and report session updates. 
        
           val sessionUpdates = events 
        
             .groupByKey(event => event.sessionId) 
        
             .mapGroupsWithState[SessionInfo, SessionUpdate](GroupStateTimeout.ProcessingTimeTimeout) { 
        
               case (sessionId: String, events: Iterator[Event], state: GroupState[SessionInfo]) => 
        
                 // If timed out, then remove session and send final update 
        
                 if (state.hasTimedOut) { 
        
                   val finalUpdate = 
        
                     SessionUpdate(sessionId, state.get.durationMs, state.get.numEvents, expired = true) 
        
                   state.remove() 
        
                   finalUpdate 
        
                 } else { 
        
                   // Update start and end timestamps in session 
        
                   val timestamps = events.map(_.timestamp.getTime).toSeq 
        
                   val updatedSession = if (state.exists) { 
        
                     val oldSession = state.get 
        
                     SessionInfo( 
        
                       oldSession.numEvents + timestamps.size, 
        
                       oldSession.startTimestampMs, 
        
                       math.max(oldSession.endTimestampMs, timestamps.max)) 
        
                   } else { 
        
                     SessionInfo(timestamps.size, timestamps.min, timestamps.max) 
        
                   } 
        
                   state.update(updatedSession) 
        
                   // Set timeout such that the session will be expired if no data received for 10 seconds 
        
                   state.setTimeoutDuration("10 seconds") 
        
                   SessionUpdate(sessionId, state.get.durationMs, state.get.numEvents, expired = false) 
        
                 } 
        
             }

(Worth noting that the code in StructuredSessionization only works with processing time. The code doesn't consider old event can update the start time of old session.)

Does this PR introduce any user-facing change?

Yes. This PR brings the new feature to support session window on both batch and streaming query, which adds a new function "session_window" which usage is similar with "window".

How was this patch tested?

New test suites. Also tested with benchmark code.

HeartSaVioR · 2021-06-25T08:33:17Z

sql/core/src/main/scala/org/apache/spark/sql/execution/aggregate/UpdatingSessionsIterator.scala

@@ -181,7 +181,8 @@ class UpdatingSessionsIterator(

  private val valueProj = GenerateUnsafeProjection.generate(valuesExpressions, inputSchema)
  private val restoreProj = GenerateUnsafeProjection.generate(inputSchema,
-    groupingExpressions.map(_.toAttribute) ++ valuesExpressions.map(_.toAttribute))
+    groupingWithoutSession.map(_.toAttribute) ++ Seq(sessionExpression.toAttribute) ++


There's a case session column is not placed at the end of the key. This line addresses such case.

HeartSaVioR · 2021-06-25T08:33:54Z

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-25T09:37:11Z

Kubernetes integration test unable to build dist.

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

SparkQA · 2021-06-25T13:08:28Z

Test build #140314 has finished for PR 33081 at commit af538b9.

This patch passes all tests.
This patch merges cleanly.
This patch adds the following public classes (experimental):
case class SessionWindow(timeColumn: Expression, gapDuration: Long) extends UnaryExpression
case class SessionWindowStateStoreRestoreExec(
case class SessionWindowStateStoreSaveExec(

SparkQA · 2021-07-14T06:22:37Z

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

SparkQA · 2021-07-14T06:56:30Z

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

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

Test build #140999 has finished for PR 33081 at commit 0af2fd1.

This patch fails Spark unit tests.
This patch merges cleanly.
This patch adds no public classes.

SparkQA · 2021-07-14T11:19:19Z

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

SparkQA · 2021-07-14T11:29:55Z

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

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

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

SparkQA · 2021-07-14T11:51:59Z

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

SparkQA · 2021-07-14T12:02:03Z

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

SparkQA · 2021-07-14T12:03:09Z

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

SparkQA · 2021-07-14T13:38:19Z

Test build #141015 has finished for PR 33081 at commit 7b74ee8.

This patch fails Spark unit tests.
This patch merges cleanly.
This patch adds no public classes.

SparkQA · 2021-07-14T14:05:41Z

Test build #141013 has finished for PR 33081 at commit 11b44bb.

This patch fails Spark unit tests.
This patch merges cleanly.
This patch adds no public classes.

SparkQA · 2021-07-14T14:40:22Z

Test build #141022 has finished for PR 33081 at commit e01a366.

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

SparkQA · 2021-07-14T15:24:59Z

Kubernetes integration test unable to build dist.

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

HeartSaVioR · 2021-07-15T02:17:17Z

cc. @viirya @xuanyuanking This is the last PR for session window feature. Please take a look. Thanks!

SparkQA · 2021-07-15T02:18:41Z

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

SparkQA · 2021-07-15T02:52:35Z

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

SparkQA · 2021-07-15T04:43:35Z

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

SparkQA · 2021-07-16T02:21:16Z

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

SparkQA · 2021-07-16T02:54:59Z

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

viirya · 2021-07-16T03:05:26Z

I will take another looks tonight.

HeartSaVioR · 2021-07-16T03:14:10Z

Rebased with master branch to see whether the pyspark lint is incurred from other PR and being fixed or not.

SparkQA · 2021-07-16T06:39:41Z

Kubernetes integration test unable to build dist.

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

SparkQA · 2021-07-16T06:48:40Z

Test build #141110 has finished for PR 33081 at commit 4a6cff3.

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

viirya · 2021-07-16T07:55:30Z

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

+    val providerOptions = Seq(
+      classOf[HDFSBackedStateStoreProvider].getCanonicalName).map { value =>
+      (SQLConf.STATE_STORE_PROVIDER_CLASS.key, value.stripSuffix("$"))


Don't we want to test against RocksDBStateStoreProvider?

Missing spot. Nice finding!

viirya

Only one question about test coverage for RocksDB. Otherwise lgtm. Thanks for the work!

SparkQA · 2021-07-16T08:32:02Z

Test build #141131 has finished for PR 33081 at commit e7a2a37.

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

SparkQA · 2021-07-16T10:51:16Z

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

SparkQA · 2021-07-16T11:25:06Z

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

HeartSaVioR · 2021-07-16T11:37:38Z

GA passed. Thanks! Merging to master/3.2!

Introduction: this PR is the last 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 proposes to support native session window. Please refer the comments/design doc in SPARK-10816 for more details on the rationalization and design (could be outdated a bit compared to the PR). The definition of the boundary of "session window" is [the timestamp of start event ~ the timestamp of last event + gap duration). That said, unlike time window, session window is a dynamic window which can expand if new input row is added to the session. To handle expansion of session window, Spark defines session window per input row, and "merge" windows if they can be merged (boundaries are overlapped). This PR leverages two different approaches on merging session windows: 1. merging session windows with Spark's aggregation logic (a variant of sort aggregation) 2. updating session window for all rows bound to the same session, and applying aggregation logic afterwards First one is preferable as it outperforms compared to the second one, though it can be only used if merging session window can be applied altogether with aggregation. It is not applicable on all the cases, so second one is used to cover the remaining cases. This PR also applies the optimization on merging input rows and existing sessions with retaining the order (group keys + start timestamp of session window), leveraging the fact the number of existing sessions per group key won't be huge. The state format is versioned, so that we can bring a new state format if we find a better one. ### Why are the changes needed? For now, to deal with sessionization, Spark requires end users to play with (flat)MapGroupsWithState directly which has a couple of major drawbacks: 1. (flat)MapGroupsWithState is lower level API and end users have to code everything in details for defining session window and merging windows 2. built-in aggregate functions cannot be used and end users have to deal with aggregation by themselves 3. (flat)MapGroupsWithState is only available in Scala/Java. With native support of session window, end users simply use "session_window" like they use "window" for tumbling/sliding window, and leverage built-in aggregate functions as well as UDAFs to simply define aggregations. Quoting the query example from test suite: ``` val inputData = MemoryStream[(String, Long)] // Split the lines into words, treat words as sessionId of events val events = inputData.toDF() .select($"_1".as("value"), $"_2".as("timestamp")) .withColumn("eventTime", $"timestamp".cast("timestamp")) .selectExpr("explode(split(value, ' ')) AS sessionId", "eventTime") .withWatermark("eventTime", "30 seconds") val sessionUpdates = events .groupBy(session_window($"eventTime", "10 seconds") as 'session, 'sessionId) .agg(count("*").as("numEvents")) .selectExpr("sessionId", "CAST(session.start AS LONG)", "CAST(session.end AS LONG)", "CAST(session.end AS LONG) - CAST(session.start AS LONG) AS durationMs", "numEvents") ``` which is same as StructuredSessionization (native session window is shorter and clearer even ignoring model classes). https://github.com/apache/spark/blob/39542bb81f8570219770bb6533c077f44f6cbd2a/examples/src/main/scala/org/apache/spark/examples/sql/streaming/StructuredSessionization.scala#L66-L105 (Worth noting that the code in StructuredSessionization only works with processing time. The code doesn't consider old event can update the start time of old session.) ### Does this PR introduce _any_ user-facing change? Yes. This PR brings the new feature to support session window on both batch and streaming query, which adds a new function "session_window" which usage is similar with "window". ### How was this patch tested? New test suites. Also tested with benchmark code. Closes #33081 from HeartSaVioR/SPARK-34893-SPARK-10816-PR-31570-part-5. Lead-authored-by: Jungtaek Lim <kabhwan.opensource@gmail.com> Co-authored-by: Liang-Chi Hsieh <viirya@gmail.com> Co-authored-by: Yuanjian Li <yuanjian.li@databricks.com> Signed-off-by: Jungtaek Lim <kabhwan.opensource@gmail.com> (cherry picked from commit f2bf8b0) Signed-off-by: Jungtaek Lim <kabhwan.opensource@gmail.com>

HeartSaVioR · 2021-07-16T11:40:16Z

Thanks all for reviewing! I merged this into master/3.2.

I filed https://issues.apache.org/jira/browse/SPARK-36172 to deal with documentation - probably need to explain about session window and how to use it in structured streaming guide doc.

SparkQA · 2021-07-16T14:59:42Z

Test build #141145 has finished for PR 33081 at commit bbade35.

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

viirya · 2021-07-16T16:09:24Z

Thank you @HeartSaVioR and @xuanyuanking!

Introduction: this PR is the last 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.) This PR proposes to support native session window. Please refer the comments/design doc in SPARK-10816 for more details on the rationalization and design (could be outdated a bit compared to the PR). The definition of the boundary of "session window" is [the timestamp of start event ~ the timestamp of last event + gap duration). That said, unlike time window, session window is a dynamic window which can expand if new input row is added to the session. To handle expansion of session window, Spark defines session window per input row, and "merge" windows if they can be merged (boundaries are overlapped). This PR leverages two different approaches on merging session windows: 1. merging session windows with Spark's aggregation logic (a variant of sort aggregation) 2. updating session window for all rows bound to the same session, and applying aggregation logic afterwards First one is preferable as it outperforms compared to the second one, though it can be only used if merging session window can be applied altogether with aggregation. It is not applicable on all the cases, so second one is used to cover the remaining cases. This PR also applies the optimization on merging input rows and existing sessions with retaining the order (group keys + start timestamp of session window), leveraging the fact the number of existing sessions per group key won't be huge. The state format is versioned, so that we can bring a new state format if we find a better one. For now, to deal with sessionization, Spark requires end users to play with (flat)MapGroupsWithState directly which has a couple of major drawbacks: 1. (flat)MapGroupsWithState is lower level API and end users have to code everything in details for defining session window and merging windows 2. built-in aggregate functions cannot be used and end users have to deal with aggregation by themselves 3. (flat)MapGroupsWithState is only available in Scala/Java. With native support of session window, end users simply use "session_window" like they use "window" for tumbling/sliding window, and leverage built-in aggregate functions as well as UDAFs to simply define aggregations. Quoting the query example from test suite: ``` val inputData = MemoryStream[(String, Long)] // Split the lines into words, treat words as sessionId of events val events = inputData.toDF() .select($"_1".as("value"), $"_2".as("timestamp")) .withColumn("eventTime", $"timestamp".cast("timestamp")) .selectExpr("explode(split(value, ' ')) AS sessionId", "eventTime") .withWatermark("eventTime", "30 seconds") val sessionUpdates = events .groupBy(session_window($"eventTime", "10 seconds") as 'session, 'sessionId) .agg(count("*").as("numEvents")) .selectExpr("sessionId", "CAST(session.start AS LONG)", "CAST(session.end AS LONG)", "CAST(session.end AS LONG) - CAST(session.start AS LONG) AS durationMs", "numEvents") ``` which is same as StructuredSessionization (native session window is shorter and clearer even ignoring model classes). https://github.com/apache/spark/blob/39542bb81f8570219770bb6533c077f44f6cbd2a/examples/src/main/scala/org/apache/spark/examples/sql/streaming/StructuredSessionization.scala#L66-L105 (Worth noting that the code in StructuredSessionization only works with processing time. The code doesn't consider old event can update the start time of old session.) Yes. This PR brings the new feature to support session window on both batch and streaming query, which adds a new function "session_window" which usage is similar with "window". New test suites. Also tested with benchmark code. Closes apache#33081 from HeartSaVioR/SPARK-34893-SPARK-10816-PR-31570-part-5. Lead-authored-by: Jungtaek Lim <kabhwan.opensource@gmail.com> Co-authored-by: Liang-Chi Hsieh <viirya@gmail.com> Co-authored-by: Yuanjian Li <yuanjian.li@databricks.com> Signed-off-by: Jungtaek Lim <kabhwan.opensource@gmail.com>

HeartSaVioR marked this pull request as draft June 25, 2021 08:29

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

HeartSaVioR commented Jun 25, 2021

View reviewed changes

HeartSaVioR force-pushed the SPARK-34893-SPARK-10816-PR-31570-part-5 branch from af538b9 to 0af2fd1 Compare July 14, 2021 05:15

HeartSaVioR mentioned this pull request Jul 14, 2021

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

Closed

HeartSaVioR force-pushed the SPARK-34893-SPARK-10816-PR-31570-part-5 branch from 0af2fd1 to 11b44bb Compare July 14, 2021 09:53

HeartSaVioR marked this pull request as ready for review July 14, 2021 10:41

HeartSaVioR and others added 10 commits July 16, 2021 12:13

[SPARK-34893][SS] Support session window natively

551bf52

Address metrics stuff

78c6f2d

Fix a bug on UpdatingSessionsIterator which made some UT failing

79f2197

Giving co-authorship to Yuanjian Li

1359123

Giving co-authorship to Liang-Chi Hsieh

0f8ad99

fix

37c98de

Fix for Scala 2.13

441d6e5

Another fix for Scala 2.13

b9c0357

Reflect review comments

a4fa37b

Reflect review comments & fix metrics

e7a2a37

HeartSaVioR force-pushed the SPARK-34893-SPARK-10816-PR-31570-part-5 branch from 4a6cff3 to e7a2a37 Compare July 16, 2021 03:13

viirya reviewed Jul 16, 2021

View reviewed changes

viirya approved these changes Jul 16, 2021

View reviewed changes

Reflect review comment

bbade35

HeartSaVioR closed this in f2bf8b0 Jul 16, 2021

	// Split the lines into words, treat words as sessionId of events
	val events = lines
	.as[(String, Timestamp)]
	.flatMap { case (line, timestamp) =>
	line.split(" ").map(word => Event(sessionId = word, timestamp))
	}

	// Sessionize the events. Track number of events, start and end timestamps of session,
	// and report session updates.
	val sessionUpdates = events
	.groupByKey(event => event.sessionId)
	.mapGroupsWithState[SessionInfo, SessionUpdate](GroupStateTimeout.ProcessingTimeTimeout) {

	case (sessionId: String, events: Iterator[Event], state: GroupState[SessionInfo]) =>

	// If timed out, then remove session and send final update
	if (state.hasTimedOut) {
	val finalUpdate =
	SessionUpdate(sessionId, state.get.durationMs, state.get.numEvents, expired = true)
	state.remove()
	finalUpdate
	} else {
	// Update start and end timestamps in session
	val timestamps = events.map(_.timestamp.getTime).toSeq
	val updatedSession = if (state.exists) {
	val oldSession = state.get
	SessionInfo(
	oldSession.numEvents + timestamps.size,
	oldSession.startTimestampMs,
	math.max(oldSession.endTimestampMs, timestamps.max))
	} else {
	SessionInfo(timestamps.size, timestamps.min, timestamps.max)
	}
	state.update(updatedSession)

	// Set timeout such that the session will be expired if no data received for 10 seconds
	state.setTimeoutDuration("10 seconds")
	SessionUpdate(sessionId, state.get.durationMs, state.get.numEvents, expired = false)
	}
	}

[SPARK-34893][SS] Support session window natively #33081

[SPARK-34893][SS] Support session window natively #33081

Conversation

HeartSaVioR commented Jun 25, 2021 • edited

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 Jun 25, 2021

Choose a reason for hiding this comment

HeartSaVioR commented Jun 25, 2021

SparkQA commented Jun 25, 2021

SparkQA commented Jun 25, 2021

SparkQA commented Jul 14, 2021

SparkQA commented Jul 14, 2021

SparkQA commented Jul 14, 2021

SparkQA commented Jul 14, 2021

SparkQA commented Jul 14, 2021

SparkQA commented Jul 14, 2021

SparkQA commented Jul 14, 2021

SparkQA commented Jul 14, 2021

SparkQA commented Jul 14, 2021

SparkQA commented Jul 14, 2021

SparkQA commented Jul 14, 2021

SparkQA commented Jul 14, 2021

SparkQA commented Jul 14, 2021

HeartSaVioR commented Jul 15, 2021

SparkQA commented Jul 15, 2021

SparkQA commented Jul 15, 2021

SparkQA commented Jul 15, 2021

SparkQA commented Jul 16, 2021

SparkQA commented Jul 16, 2021

viirya commented Jul 16, 2021

HeartSaVioR commented Jul 16, 2021

SparkQA commented Jul 16, 2021

SparkQA commented Jul 16, 2021

viirya Jul 16, 2021

Choose a reason for hiding this comment

HeartSaVioR Jul 16, 2021

Choose a reason for hiding this comment

viirya left a comment

Choose a reason for hiding this comment

SparkQA commented Jul 16, 2021

SparkQA commented Jul 16, 2021

SparkQA commented Jul 16, 2021

HeartSaVioR commented Jul 16, 2021

HeartSaVioR commented Jul 16, 2021

SparkQA commented Jul 16, 2021

viirya commented Jul 16, 2021

HeartSaVioR commented Jun 25, 2021 •

edited