[SPARK-28177][SQL] Adjust post shuffle partition number in adaptive execution #24978

carsonwang · 2019-06-26T22:42:19Z

What changes were proposed in this pull request?

This is to implement a ReduceNumShufflePartitions rule in the new adaptive execution framework introduced in #24706. This rule is used to adjust the post shuffle partitions based on the map output statistics.

How was this patch tested?

Added ReduceNumShufflePartitionsSuite

gatorsmile · 2019-06-26T22:47:43Z

cc @hvanhovell @maryannxue @cloud-fan

SparkQA · 2019-06-27T01:44:46Z

Test build #106949 has finished for PR 24978 at commit 6f6bdaf.

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

cloud-fan · 2019-06-27T06:46:57Z

sql/catalyst/src/main/scala/org/apache/spark/sql/internal/SQLConf.scala

-      .doc("When true, enable runtime query re-optimization.")
-      .booleanConf
-      .createWithDefault(false)
-
  val ADAPTIVE_EXECUTION_ENABLED = buildConf("spark.sql.adaptive.enabled")


This is an umbrella config, let's create a new config for ReduceNumShufflePartitions. How about spark.sql.adaptive.reducePostShufflePartitions.enabled?

+1. Being a bit pedantic here, but let's try to do orthogonal changes like removing the RUNTIME_REOPTIMIZATION_ENABLED flag in separate PRs. That makes the cognitive load of reviewing lower, and it also improves git blame.

Agree to add a config for ReduceNumShufflePartitions. I was removing RUNTIME_REOPTIMIZATION_ENABLED because RUNTIME_REOPTIMIZATION_ENABLED and ADAPTIVE_EXECUTION_ENABLED look like the same config now. I didn't see a way to enable ReduceNumShufflePartitions but disable runtime reoptimization.

The argument of not changing RUNTIME_REOPTIMIZATION_ENABLED to ADAPTIVE_EXECUTION_ENABLED is just to keep this PR independent and smaller.

Whatever name it is, we should have an umbrella flag that turns everything off. We could add another flag, though, just to turn off the SMJ -> BHJ runtime optimization, although we might have to do it in a slightly trickier way given that this opt comes naturally with the framework itself.

All right, I will use RUNTIME_REOPTIMIZATION_ENABLED as the umbrella flag in this PR.

hvanhovell · 2019-06-27T07:50:16Z

sql/core/src/main/scala/org/apache/spark/sql/execution/exchange/ShuffleExchangeExec.scala

-    var newPartitioning: Partitioning,
-    child: SparkPlan,
-    @transient coordinator: Option[ExchangeCoordinator]) extends Exchange {
+    desiredPartitioning: Partitioning,


outputPartitioning?

hvanhovell · 2019-06-27T07:56:04Z

sql/catalyst/src/main/scala/org/apache/spark/sql/internal/SQLConf.scala


  def minNumPostShufflePartitions: Int =
    getConf(SHUFFLE_MIN_NUM_POSTSHUFFLE_PARTITIONS)

+  def maxNumPostShufflePartitions: Int =
+    getConf(SHUFFLE_MAX_NUM_POSTSHUFFLE_PARTITIONS).getOrElse(numShufflePartitions)


We may want to set the default value here to something much higher (Int.Max?). The upper bound here is determined by the shuffles for which we are trying to reduce the number of partitions.

We actually use this as the initial shuffle partition number, which will be set in the ShuffleExchangeExec. So this is expected to be a reasonable value, instead of Int.Max. This seems to be a little misleading if a user is not aware it is used as the initial shuffle partition number.

Why would we have two confs for adaptive and non-adaptive execution?

People usually already have tuned spark.sql.shuffle.partitions for each workload they run periodically. The initial shuffle partition number for adaptive execution allows user to set a global value (relative large) that can work for all queries. If user doesn't set it, it will default to spark.sql.shuffle.partitions.

hvanhovell · 2019-06-27T08:00:23Z

...core/src/main/scala/org/apache/spark/sql/execution/adaptive/ReduceNumShufflePartitions.scala

+        ThreadUtils.awaitResult(metricsFuture, Duration.Zero)
+    }
+
+    if (!plan.collectLeaves().forall(_.isInstanceOf[QueryStageExec])) {


How about first collecting all the leaves, then checking if they are all QueryStageExec nodes, and if they are collect the map stats outputs? That seems more efficient.

Good point!

JkSelf · 2019-06-27T08:07:41Z

...core/src/main/scala/org/apache/spark/sql/execution/adaptive/ReduceNumShufflePartitions.scala

+      case ReusedQueryStageExec(_, stage: ShuffleQueryStageExec, _) =>
+        val metricsFuture = stage.mapOutputStatisticsFuture
+        assert(metricsFuture.isCompleted, "ShuffleQueryStageExec should already be ready")
+        ThreadUtils.awaitResult(metricsFuture, Duration.Zero)


Here, the shuffle query stage is already ready, why we still call the ThreadUtils.awaitResult method?

because we need to get the value from a Future...

Got it. Thanks for your explanation.

future.get()?

It is a scala Future and future.get() seems to be only available for Java Future.

JkSelf · 2019-06-27T08:07:48Z

...core/src/main/scala/org/apache/spark/sql/execution/adaptive/ReduceNumShufflePartitions.scala

+        ThreadUtils.awaitResult(metricsFuture, Duration.Zero)
+    }
+
+    if (!plan.collectLeaves().forall(_.isInstanceOf[QueryStageExec])) {


When the leaf node is BroadcastQueryStageExec, Do we need to adjust the reduce number ?

We are ignoring BroadcastQueryStageExec right?

We can have situations with BHJ where:

The BHJ is from compile-time and one side is a broadcast stage;

The BHJ is from a former AQE optimization, then one side is shuffle stage and the other is a broadcast stage.

I believe scenario 2 may still be applicable here, so maybe we can change the condition to " isAllQueryStage && shuffleStageCount > 0"?

Obviously we don't want to adjust num shuffle partitions if there is no shuffle in this stage, so +1 to change the condition to isAllQueryStage && shuffleStageCount > 0.

I believe we are safe here. After checking isAllQueryStage, we will get shuffle metrics from shuffle stages and we will only adjust num shuffle partitions if num shuffle metrics > 0.

JkSelf · 2019-06-27T08:12:59Z

...core/src/main/scala/org/apache/spark/sql/execution/adaptive/ReduceNumShufflePartitions.scala

+      // we should skip it when calculating the `partitionStartIndices`.
+      val validMetrics = shuffleMetrics.filter(_ != null)
+      if (validMetrics.nonEmpty) {
+        val partitionStartIndices = estimatePartitionStartIndices(validMetrics.toArray)


We may need add a check of whether all the pre-shuffle partitions is same before calling the estimatePartitionStartIndices method to avoid the assert exception in Line 130.

Good catch. I'll add the check and the unit test.

maryannxue · 2019-06-27T21:48:02Z

...core/src/main/scala/org/apache/spark/sql/execution/adaptive/ReduceNumShufflePartitions.scala

+          // number of output partitions.
+          case stage: ShuffleQueryStageExec =>
+            CoalescedShuffleReaderExec(stage, partitionStartIndices)
+          case reused: ReusedQueryStageExec if reused.plan.isInstanceOf[ShuffleQueryStageExec] =>


Can we use pattern match here as in other places too?
Alternatively, as we already have BroadcastQueryStageExec.isBroadcastQueryStageExec, we could accordingly add ShuffleQueryStageExec.isShuffleQueryStageExec and replace all such occurrences with just one "case".

SparkQA · 2019-06-29T21:22:11Z

Test build #107043 has finished for PR 24978 at commit 228ed58.

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

SparkQA · 2019-06-29T21:43:38Z

Test build #107045 has finished for PR 24978 at commit 7f0397a.

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

SparkQA · 2019-06-30T07:05:02Z

Test build #107047 has finished for PR 24978 at commit b767977.

This patch fails due to an unknown error code, -9.
This patch merges cleanly.
This patch adds no public classes.

carsonwang · 2019-06-30T17:59:47Z

test this please

SparkQA · 2019-06-30T21:11:07Z

Test build #107053 has finished for PR 24978 at commit b767977.

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

cloud-fan · 2019-07-01T05:41:17Z

sql/core/src/main/scala/org/apache/spark/sql/execution/exchange/EnsureRequirements.scala

-  }
+  private def defaultNumPreShufflePartitions: Int =
+    if (conf.adaptiveExecutionEnabled) {
+      conf.maxNumPostShufflePartitions


We should mention it in the config doc that this config is used as the initial num pre shuffle partitions.

Yes, added that.

cloud-fan · 2019-07-01T05:43:58Z

LGTM

maryannxue · 2019-07-01T19:40:40Z

Some general comments:

Can we remove the config renaming from this PR as @hvanhovell requested? We can have a separate PR to address this. And also I can file a follow-up PR to add an independent conf for SMJ -> BHJ.
Can we add end2end tests where SMJ->BHJ and reduce-shuffle-num happen at the same time?

JkSelf · 2019-07-02T03:08:17Z

@carsonwang @maryannxue I encountered the following exception with q1 of TPC-DS in yarn mode when enable or disable spark.sql.adaptive.reducePostShufflePartitions.enabled . And this exception may be not related with this PR, but with PR#24706
org.apache.spark.SparkException: Adaptive execution failed due to stage materialization failures. at org.apache.spark.sql.execution.adaptive.AdaptiveSparkPlanExec.cleanUpAndThrowException(AdaptiveSparkPlanExec.scala:383) at org.apache.spark.sql.execution.adaptive.AdaptiveSparkPlanExec.doExecute(AdaptiveSparkPlanExec.scala:158) at org.apache.spark.sql.execution.SparkPlan.$anonfun$execute$1(SparkPlan.scala:168) at org.apache.spark.sql.execution.SparkPlan.$anonfun$executeQuery$1(SparkPlan.scala:192) at org.apache.spark.rdd.RDDOperationScope$.withScope(RDDOperationScope.scala:151) at org.apache.spark.sql.execution.SparkPlan.executeQuery(SparkPlan.scala:189) at org.apache.spark.sql.execution.SparkPlan.execute(SparkPlan.scala:164) at org.apache.spark.sql.execution.SparkPlan.getByteArrayRdd(SparkPlan.scala:284) at org.apache.spark.sql.execution.SparkPlan.executeCollect(SparkPlan.scala:333) at org.apache.spark.sql.execution.SparkPlan.executeCollectPublic(SparkPlan.scala:363) at org.apache.spark.sql.execution.HiveResult$.hiveResultString(HiveResult.scala:52) at org.apache.spark.sql.hive.thriftserver.SparkSQLDriver.$anonfun$run$1(SparkSQLDriver.scala:65) at org.apache.spark.sql.execution.SQLExecution$.$anonfun$withNewExecutionId$4(SQLExecution.scala:100) at org.apache.spark.sql.execution.SQLExecution$.withSQLConfPropagated(SQLExecution.scala:160) at org.apache.spark.sql.execution.SQLExecution$.withNewExecutionId(SQLExecution.scala:87) at org.apache.spark.sql.hive.thriftserver.SparkSQLDriver.run(SparkSQLDriver.scala:65) at org.apache.spark.sql.hive.thriftserver.SparkSQLCLIDriver.processCmd(SparkSQLCLIDriver.scala:367) at org.apache.hadoop.hive.cli.CliDriver.processLine(CliDriver.java:376) at org.apache.hadoop.hive.cli.CliDriver.processLine(CliDriver.java:311) at org.apache.hadoop.hive.cli.CliDriver.processReader(CliDriver.java:409) at org.apache.hadoop.hive.cli.CliDriver.processFile(CliDriver.java:425) at org.apache.spark.sql.hive.thriftserver.SparkSQLCLIDriver$.main(SparkSQLCLIDriver.scala:196) at org.apache.spark.sql.hive.thriftserver.SparkSQLCLIDriver.main(SparkSQLCLIDriver.scala) at sun.reflect.NativeMethodAccessorImpl.invoke0(Native Method) at sun.reflect.NativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.java:62) at sun.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43) at java.lang.reflect.Method.invoke(Method.java:498) at org.apache.spark.deploy.JavaMainApplication.start(SparkApplication.scala:52) at org.apache.spark.deploy.SparkSubmit.org$apache$spark$deploy$SparkSubmit$$runMain(SparkSubmit.scala:920) at org.apache.spark.deploy.SparkSubmit.doRunMain$1(SparkSubmit.scala:179) at org.apache.spark.deploy.SparkSubmit.submit(SparkSubmit.scala:202) at org.apache.spark.deploy.SparkSubmit.doSubmit(SparkSubmit.scala:89) at org.apache.spark.deploy.SparkSubmit$$anon$2.doSubmit(SparkSubmit.scala:999) at org.apache.spark.deploy.SparkSubmit$.main(SparkSubmit.scala:1008) at org.apache.spark.deploy.SparkSubmit.main(SparkSubmit.scala) Caused by: org.apache.spark.SparkException: Failed to materialize query stage: ShuffleQueryStage 6

JkSelf · 2019-07-02T03:10:35Z

The full log info is in google doc.

gcz2022 · 2019-07-02T03:33:15Z

...core/src/main/scala/org/apache/spark/sql/execution/adaptive/ReduceNumShufflePartitions.scala

+      return plan
+    }
+    if (!plan.collectLeaves().forall(_.isInstanceOf[QueryStageExec])) {
+      // If not all leaf nodes are query stages, it's not safe to reduce the number of


Does this mean the case like https://github.com/Intel-bigdata/spark-adaptive/pull/54 ?
:nit Maybe we can say If not all leaf nodes are query stages, for example when some depending stage is already bucketed, there isn't a shuffle or query stage for that branch to be clear because normally all leaves are QS.

Yes, that case is covered. There are many cases the leaves are not query stage, for example in the stage that does a table scan, the leaf is a table scan operator. The comment seems to cover these.

gcz2022 · 2019-07-02T03:48:19Z

...core/src/main/scala/org/apache/spark/sql/execution/adaptive/ReduceNumShufflePartitions.scala

+      val distinctNumPreShufflePartitions =
+        validMetrics.map(stats => stats.bytesByPartitionId.length).distinct
+
+      if (validMetrics.nonEmpty && distinctNumPreShufflePartitions.length == 1) {


If a user calls val datasetAfter = datasetBefore.repartition(300), then he wants the post-shuffle partitionNum to be 300 exactly. But in this case we will go inside this branch, and then the partition merging on reduce side will break the user's expectation? cc @cloud-fan
Should we check the original operation, if it's repartition we don't do Adaptive Execution then?

This was also discussed in the original PR. For df.repartition(500), AE does use the specified number for repartition. That is, each stage will writes the map output with 500 partitions. However, in the following up stage, AE may launch tasks less than 500 as one task can process multiple continues blocks. I think we can still do adaptive execution when it is possible. We can add a option to disable AE for repartition later if that is necessary.

It's pretty good to add a config entry, I think it's the best to make it internal(hidden). At the same time, we can add a function called repartitonWithAdvice, the different between it and repartition is whether the newly added config entry is set to true.

RepartitionWithAdvice means enabling AdaptiveExecution, to be different from the accurate repartition.

Do we have this problem in ExchangeCoordinator? At physical phase we have no idea if the shuffle comes from a df.repartition or is added by EnsureRequirements. We may need to add a boolean flag in ShuffleExchangeExec to indicate that this shuffle can't change its num partitions. Maybe we can fix it in a followup.

gcz2022 · 2019-07-02T03:51:10Z

...core/src/main/scala/org/apache/spark/sql/execution/adaptive/ReduceNumShufflePartitions.scala

+      // We may get different pre-shuffle partition number if user calls repartition manually.
+      // We don't reduce shuffle partition number in that case.
+      val distinctNumPreShufflePartitions =
+        validMetrics.map(stats => stats.bytesByPartitionId.length).distinct


There are 2 conditions we don't do Adaptive Execution, so plan is directly returned.
Maybe better to aggregate these conditions?

What do you mean by aggregate ?

Never mind~

gcz2022 · 2019-07-02T04:15:03Z

@maryannxue I am curious about why do we need a separate switch for SMJ->BHJ? In the new adaptive execution with re-optimization logic, there can be tons of switches we can add...

cloud-fan · 2019-07-02T08:17:42Z

@JkSelf did you use external shuffle service? master branch is incompatible with external shuffle service 2.x

JkSelf · 2019-07-02T08:28:54Z

@cloud-fan Thanks for your quick reply. Yes, I used the external shuffle service. And the above exception is gone when disable external shuffle service. I am curious about why the master branch is incompatible with external shuffle service? Thanks.

cloud-fan · 2019-07-02T09:16:23Z

@JkSelf see #24565

hvanhovell · 2019-07-02T09:46:13Z

...core/src/main/scala/org/apache/spark/sql/execution/adaptive/ReduceNumShufflePartitions.scala

+      // have same number of output partitions.
+      plan
+    } else {
+      val shuffleStages = plan.collect {


This does not work well with Union right? Is there a way we can identify subtrees in the stage where all the input partitions are the same?

Right, Union may have child stages with different partition number, see the unit test added. Currently we check the shuffle metrics to ensure the pre-shuffle partition number of the stages are the same.

This can get pretty complicated if we have multiple Union and Join in one stage. For now I'm OK with this simple approach that we always require all shuffles in one stage have the same number of partitions. BTW this is an existing issue.

weixiuli · 2019-07-02T11:24:13Z

...core/src/main/scala/org/apache/spark/sql/execution/adaptive/ReduceNumShufflePartitions.scala

+
+  override def apply(plan: SparkPlan): SparkPlan = {
+    if (!conf.reducePostShufflePartitionsEnabled) {
+      return plan


return is not needed.

Without return, I'll need a else with more indentation. I saw a few rules write like this, so I was just copying that.

weixiuli · 2019-07-02T11:35:24Z

...core/src/main/scala/org/apache/spark/sql/execution/adaptive/ReduceNumShufflePartitions.scala

+        nextShuffleInputSize += mapOutputStatistics(j).bytesByPartitionId(i)
+        j += 1
+      }
+


var nextShuffleInputSize = 0L var j = 0 while (j < mapOutputStatistics.length) { nextShuffleInputSize += mapOutputStatistics(j).bytesByPartitionId(i) j += 1 }

=>
val nextShuffleInputSize = mapOutputStatistics.map(_.bytesByPartitionId(i)).sum

This code was original there in ExchangeCoordinator. Not sure if it is in purpose to avoid creating many additional arrays when we write as you proposed. So let's leave it as is.

carsonwang · 2019-07-03T00:00:14Z

sql/core/src/test/scala/org/apache/spark/sql/execution/adaptive/AdaptiveQueryExecSuite.scala

@@ -92,6 +94,30 @@ class AdaptiveQueryExecSuite extends QueryTest with SharedSQLContext {
    }
  }

+  test("Change merge join to broadcast join and reduce number of shuffle partitions") {


@maryannxue, added an end2end test where SMJ->BHJ and reduce-shuffle-num happen at the same time . Actually REDUCE_POST_SHUFFLE_PARTITIONS_ENABLED is enabled by default, so other test cases also reduce shuffle partition numbers if we don't disable it. But that doesn't seem to be a problem.

SparkQA · 2019-07-03T02:48:09Z

Test build #107133 has finished for PR 24978 at commit 65ea8f6.

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

SparkQA · 2019-07-03T03:17:26Z

Test build #107137 has finished for PR 24978 at commit 27a8180.

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

carsonwang · 2019-07-03T05:36:46Z

test this please

SparkQA · 2019-07-03T07:05:02Z

Test build #107157 has finished for PR 24978 at commit 27a8180.

This patch fails due to an unknown error code, -9.
This patch merges cleanly.
This patch adds no public classes.

maryannxue · 2019-07-03T15:34:07Z

sql/core/src/test/scala/org/apache/spark/sql/execution/adaptive/AdaptiveQueryExecSuite.scala

@@ -82,6 +83,7 @@ class AdaptiveQueryExecSuite extends QueryTest with SharedSQLContext {
  test("Change merge join to broadcast join") {
    withSQLConf(
        SQLConf.RUNTIME_REOPTIMIZATION_ENABLED.key -> "true",
+        SQLConf.REDUCE_POST_SHUFFLE_PARTITIONS_ENABLED.key -> "false",


Any reason for this?

I added the same test with REDUCE_POST_SHUFFLE_PARTITIONS_ENABLED set to true and checked the post shuffle partition. So I set it to false in this test.

I don't think it's necessary, but if you do wanna test it with "false", how about do a combo: Seq("true", "false").foreach?

Ok, let me just remove this for now.

carsonwang · 2019-07-03T17:25:58Z

Test this please

SparkQA · 2019-07-04T03:46:23Z

Test build #107206 has finished for PR 24978 at commit 4b001ca.

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

cloud-fan · 2019-07-04T08:03:17Z

thanks, merging to master!

carsonwang · 2019-07-08T06:27:36Z

Thank you all!

…xecution This is to implement a ReduceNumShufflePartitions rule in the new adaptive execution framework introduced in apache#24706. This rule is used to adjust the post shuffle partitions based on the map output statistics. Added ReduceNumShufflePartitionsSuite Closes apache#24978 from carsonwang/reduceNumShufflePartitions. Authored-by: Carson Wang <carson.wang@intel.com> Signed-off-by: Wenchen Fan <wenchen@databricks.com>

carsonwang added 3 commits June 26, 2019 06:07

add ReduceNumShufflePartitions rule

31756cb

Fix tests

54c8715

use spark.sql.adaptive.enabled

6f6bdaf

cloud-fan reviewed Jun 27, 2019

View reviewed changes

hvanhovell reviewed Jun 27, 2019

View reviewed changes

JkSelf reviewed Jun 27, 2019

View reviewed changes

maryannxue reviewed Jun 27, 2019

View reviewed changes

dongjoon-hyun added the SQL label Jun 28, 2019

carsonwang added 2 commits June 30, 2019 03:08

address comments

228ed58

tweak

7f0397a

fix

b767977

cloud-fan reviewed Jul 1, 2019

View reviewed changes

gcz2022 reviewed Jul 2, 2019

View reviewed changes

hvanhovell reviewed Jul 2, 2019

View reviewed changes

weixiuli reviewed Jul 2, 2019

View reviewed changes

carsonwang added 2 commits July 3, 2019 07:31

use runtime reoptimization config and add a test

65ea8f6

tweak

27a8180

carsonwang commented Jul 3, 2019

View reviewed changes

maryannxue reviewed Jul 3, 2019

View reviewed changes

update test

4b001ca

cloud-fan closed this in cec6a32 Jul 4, 2019

carsonwang mentioned this pull request Jul 10, 2019

[SPARK-28339][SQL] Rename Spark SQL adaptive execution configuration name #25102

Closed

[SPARK-28177][SQL] Adjust post shuffle partition number in adaptive execution #24978

[SPARK-28177][SQL] Adjust post shuffle partition number in adaptive execution #24978

Conversation

carsonwang commented Jun 26, 2019

What changes were proposed in this pull request?

How was this patch tested?

gatorsmile commented Jun 26, 2019

SparkQA commented Jun 27, 2019

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

cloud-fan Jun 27, 2019 • edited Loading

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

hvanhovell Jun 27, 2019 • edited Loading

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

SparkQA commented Jun 29, 2019

SparkQA commented Jun 29, 2019

SparkQA commented Jun 30, 2019

carsonwang commented Jun 30, 2019

SparkQA commented Jun 30, 2019

Choose a reason for hiding this comment

Choose a reason for hiding this comment

cloud-fan commented Jul 1, 2019

maryannxue commented Jul 1, 2019

JkSelf commented Jul 2, 2019

JkSelf commented Jul 2, 2019

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

gcz2022 Jul 3, 2019 • edited Loading

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

gcz2022 commented Jul 2, 2019

cloud-fan commented Jul 2, 2019

JkSelf commented Jul 2, 2019

cloud-fan commented Jul 2, 2019

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

SparkQA commented Jul 3, 2019

SparkQA commented Jul 3, 2019

carsonwang commented Jul 3, 2019

SparkQA commented Jul 3, 2019

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

carsonwang commented Jul 3, 2019

cloud-fan Jun 27, 2019 •

edited

Loading

hvanhovell Jun 27, 2019 •

edited

Loading

gcz2022 Jul 3, 2019 •

edited

Loading