[BEAM-10940] Add sdf initiated checkpoint support to portable Flink. #13105

boyuanzz · 2020-10-14T04:09:32Z

This PR added the sdf self-checkpoint support to portable Flink by using timer and state where

In batch, we use InMemoryTimerInternals and InMemoryStateInternals to set the timer and persist residual elements.
In streaming, we use SdfFlinkTimerInternals as the wrapper of FlinkTimerInternals and SdfFlinkStateInternals as the wrapper of FlinkStateInternals for one specific key.

For more design details, please refer to this doc

r: @robertwb @mxm
cc: @ibzib @lukecwik

Thank you for your contribution! Follow this checklist to help us incorporate your contribution quickly and easily:

Choose reviewer(s) and mention them in a comment (R: @username).
Format the pull request title like [BEAM-XXX] Fixes bug in ApproximateQuantiles, where you replace BEAM-XXX with the appropriate JIRA issue, if applicable. This will automatically link the pull request to the issue.
Update CHANGES.md with noteworthy changes.
If this contribution is large, please file an Apache Individual Contributor License Agreement.

See the Contributor Guide for more tips on how to make review process smoother.

Post-Commit Tests Status (on master branch)

Lang	Dataflow	Samza	Twister2
Go	---	---	---
Java
Python		---	---
XLang	---	---	---

Pre-Commit Tests Status (on master branch)

---	Java	Python	Go	Website	Whitespace	Typescript
Non-portable
Portable	---		---	---	---	---

See .test-infra/jenkins/README for trigger phrase, status and link of all Jenkins jobs.

GitHub Actions Tests Status (on master branch)

See CI.md for more information about GitHub Actions CI.

boyuanzz · 2020-10-21T00:51:02Z

Run Java Flink PortableValidatesRunner Streaming

boyuanzz · 2020-10-21T00:51:24Z

Run Java Flink PortableValidatesRunner Batch

boyuanzz · 2020-10-21T02:08:59Z

Run Java Flink PortableValidatesRunner Streaming

boyuanzz · 2020-10-21T02:09:05Z

Run Java Flink PortableValidatesRunner Batch

boyuanzz · 2020-10-21T19:49:34Z

Run Java Flink PortableValidatesRunner Batch

boyuanzz · 2020-10-21T19:49:42Z

Run Java Flink PortableValidatesRunner Streaming

boyuanzz · 2020-10-21T20:09:34Z

Run Python_PVR_Flink PreCommit

boyuanzz · 2020-10-22T00:42:49Z

SDF + side input tests fail on validateParDo:

beam/runners/core-construction-java/src/main/java/org/apache/beam/runners/core/construction/graph/PipelineValidator.java

Lines 231 to 237 in ddcb600

    
           for (String sideInputId : payload.getSideInputsMap().keySet()) { 
        
             checkArgument( 
        
                 transform.containsInputs(sideInputId), 
        
                 "Transform %s side input %s is not listed in the transform's inputs", 
        
                 id, 
        
                 sideInputId); 
        
           }

boyuanzz · 2020-10-22T01:22:52Z

Run Java Flink PortableValidatesRunner Batch

boyuanzz · 2020-10-22T01:29:18Z

Run Java Flink PortableValidatesRunner Batch

boyuanzz · 2020-10-22T01:53:23Z

SDF + side input tests fail on validateParDo:

beam/runners/core-construction-java/src/main/java/org/apache/beam/runners/core/construction/graph/PipelineValidator.java

Lines 231 to 237 in ddcb600

for (String sideInputId : payload.getSideInputsMap().keySet()) {

checkArgument(

transform.containsInputs(sideInputId),

"Transform %s side input %s is not listed in the transform's inputs",

id,

sideInputId);

}

It seems like we only have main input for SplittableParDo transform:

splittableParDo.getInputsMap(): {org.apache.beam.sdk.values.PCollection.<init>:399#a00997a449310e6c=PAssert$0/GroupGlobally/Create.Values/Read(CreateSource)/ParDo(OutputSingleSource)/ParMultiDo(OutputSingleSource).output}

@lukecwik Have we added the logic to populate the side input into transform proto?

boyuanzz · 2020-10-22T18:46:13Z

Run Java Flink PortableValidatesRunner Batch

boyuanzz · 2020-10-22T18:54:38Z

Run Java Flink PortableValidatesRunner Batch

boyuanzz · 2020-10-22T19:04:46Z

Run Java Flink PortableValidatesRunner Batch

boyuanzz · 2020-10-22T19:18:07Z

Run Java Flink PortableValidatesRunner Batch

boyuanzz · 2020-10-22T19:33:40Z

I can verify that the pipeline proto is correct before fusion. The validation error happens when fusion returns the new pipeline proto. It seems like the something is wrong in pipeline fuser.

boyuanzz · 2020-10-23T00:09:53Z

Run Java Flink PortableValidatesRunner Batch

boyuanzz · 2020-10-23T02:15:08Z

Run Java Flink PortableValidatesRunner Batch

boyuanzz · 2020-10-23T02:15:25Z

Run Java Flink PortableValidatesRunner Streaming

boyuanzz · 2020-10-23T17:57:46Z

Run Java Flink PortableValidatesRunner Batch

boyuanzz · 2020-10-23T17:57:54Z

Run Java Flink PortableValidatesRunner Streaming

boyuanzz · 2020-10-23T20:00:21Z

Run Java Flink PortableValidatesRunner Batch

boyuanzz · 2020-10-23T20:00:28Z

Run Java Flink PortableValidatesRunner Streaming

boyuanzz · 2020-10-23T23:42:19Z

The testPairWithIndexWindowedTimestampedBounded in batch also fails without my changes(#13176, https://ci-beam.apache.org/job/beam_PostCommit_Java_PVR_Flink_Batch_PR/183/testReport/org.apache.beam.sdk.transforms/SplittableDoFnTest/).

boyuanzz · 2020-10-26T17:16:11Z

Run Portable_Python PreCommit

boyuanzz · 2020-10-26T17:49:11Z

Run Java Flink PortableValidatesRunner Streaming

boyuanzz · 2020-10-26T22:03:07Z

Run Java Flink PortableValidatesRunner Batch

boyuanzz · 2020-10-26T22:03:16Z

Run Java Flink PortableValidatesRunner Streaming

boyuanzz · 2020-10-26T22:47:03Z

Run Java PreCommit

boyuanzz · 2020-10-26T22:47:12Z

Run Python_PVR_Flink PreCommit

boyuanzz · 2020-10-27T00:17:56Z

Run Java Flink PortableValidatesRunner Streaming

mxm · 2020-10-27T09:13:09Z

runners/flink/job-server/flink_job_server.gradle

      }
    },
    testFilter: {
      // TODO(BEAM-10016)
      excludeTestsMatching 'org.apache.beam.sdk.transforms.FlattenTest.testFlattenWithDifferentInputAndOutputCoders2'
+      excludeTestsMatching 'org.apache.beam.sdk.transforms.SplittableDoFnTest.testPairWithIndexWindowedTimestampedBounded'


Maybe add a comment why this one is excluded?

I haven't explore the failure yet but the test fails on the same reason(wrong results) for batch and streaming. I filed a jira for tracking this: https://issues.apache.org/jira/browse/BEAM-11141

I figured out the test failure and enabled the test in the latest commit.

mxm · 2020-10-27T09:17:34Z

...src/main/java/org/apache/beam/runners/flink/translation/wrappers/streaming/DoFnOperator.java

@@ -1315,7 +1316,7 @@ void processPendingProcessingTimeTimers() {
        keyedStateBackend.setCurrentKey(internalTimer.getKey());
        TimerData timer = internalTimer.getNamespace();
        checkInvokeStartBundle();
-        fireTimer(timer);
+        fireTimerInternal((ByteBuffer) internalTimer.getKey(), timer);


I'm assuming this was a bug?

I think so. The only difference is that fireTimerInternal will grab the lock and set the key for state backend, which is helpful when the state is needed. Another potential bug around here is that Flink operator will clean up global state first then fire the timer, which may lead to data loss if firing timer is for retrieving the state.

mxm · 2020-10-27T09:19:20Z

...src/main/java/org/apache/beam/runners/flink/translation/wrappers/streaming/DoFnOperator.java

+    /** Holds the watermark when there is an sdf timer. */
+    private void onNewSdfTimer(TimerData newTimer) {
+      Preconditions.checkState(
+          StateAndTimerBundleCheckpointHandler.isSdfTimer(newTimer.getTimerId()));
+      Preconditions.checkState(timerUsesOutputTimestamp(newTimer));
+      keyedStateInternals.addWatermarkHoldUsage(newTimer.getOutputTimestamp());
+    }


Wouldn't it make sense to integrate this check with the timerUsesOutputTimestamp method?

We could rename the mentioned method to onFiredTimer and include the checks for output watermark holds in there.

Sounds good.

Sorry I want to revisit the idea of having onFiredTimer here. I think it's a good idea to have onFiredTimer for firing timers. But the function onNewSdfTimer and onNewEventTimer is about to set watermark hold when registering timers. Different from event timer, an SDF timer must have the output timestamp for controlling watermark hold. It's important for SDF execution. That's why we have a check instead of an if block here.

mxm · 2020-10-27T09:19:52Z

...src/main/java/org/apache/beam/runners/flink/translation/wrappers/streaming/DoFnOperator.java

+    private void onRemovedSdfTimer(TimerData removedTimer) {
+      Preconditions.checkState(
+          StateAndTimerBundleCheckpointHandler.isSdfTimer(removedTimer.getTimerId()));
+      Preconditions.checkState(timerUsesOutputTimestamp(removedTimer));
+      // Remove the watermark hold which is set for this sdf timer.
+      keyedStateInternals.removeWatermarkHoldUsage(removedTimer.getOutputTimestamp());
+    }


Wouldn't it make sense to integrate this check with the timerUsesOutputTimestamp method?

We could rename the mentioned method to onFiredTimer and include the checks for output watermark holds in there.

mxm · 2020-10-27T09:22:07Z

...src/main/java/org/apache/beam/runners/flink/translation/wrappers/streaming/DoFnOperator.java

+          if (StateAndTimerBundleCheckpointHandler.isSdfTimer(timer.getTimerId())) {
+            onNewSdfTimer(timer);
+          }


Could be simplified by having a generic call here to onFiredTimer. See above.

mxm · 2020-10-27T09:22:14Z

...src/main/java/org/apache/beam/runners/flink/translation/wrappers/streaming/DoFnOperator.java

        if (timer.getDomain() == TimeDomain.EVENT_TIME) {
          onRemovedEventTimer(timer);
+        } else if (StateAndTimerBundleCheckpointHandler.isSdfTimer(timer.getTimerId())) {
+          onRemovedSdfTimer(timer);


Could be simplified by having a generic call here to onFiredTimer. See above.

mxm · 2020-10-27T09:25:02Z

...rg/apache/beam/runners/flink/translation/wrappers/streaming/ExecutableStageDoFnOperator.java

+  }
+
+  /** A {@link StateInternals} for keeping {@link DelayedBundleApplication}s as states. */
+  class SdfFlinkStateInternals implements StateInternals {


Maybe we should move these out of this class.

I want to keep SdfFlinkInternals and SdfTimerInternals as inner class so that they can access to getKeyedStateBackend(), timerInternals and stateInternals from outer class.

mxm · 2020-10-27T09:25:11Z

...rg/apache/beam/runners/flink/translation/wrappers/streaming/ExecutableStageDoFnOperator.java

+   * StateAndTimerBundleCheckpointHandler} to handle {@link
+   * org.apache.beam.model.fnexecution.v1.BeamFnApi.DelayedBundleApplication}.
+   */
+  class SdfFlinkTimerInternalsFactory implements TimerInternalsFactory<InputT> {


Maybe we should move these out of this class.

mxm · 2020-10-27T09:47:40Z

...nk/src/main/java/org/apache/beam/runners/flink/FlinkStreamingPortablePipelineTranslator.java

+        // For an SDF, we know that the input element should be
+        // KV<KV<element, KV<restriction, watermarkState>>, size>. We are going to use the element
+        // as the key.
+        if (!(((KvCoder) valueCoder).getKeyCoder() instanceof KvCoder)) {
+          throw new IllegalStateException(
+              String.format(
+                  Locale.ENGLISH,
+                  "The element coder for splittable DoFn '%s' must be KVCoder(KvCoder, DoubleCoder) but is: %s",
+                  inputPCollectionId,
+                  valueCoder.getClass().getSimpleName()));
+        }
+        keyCoder = ((KvCoder) ((KvCoder) valueCoder).getKeyCoder()).getKeyCoder();
+        keySelector = new SdfByteBufferKeySelector(keyCoder);
+      }


Will SDFs ever support stateful operations? If so, this wouldn't work anymore because keys are not guaranteed to be processed on the same operator instance.

At least for now we don't support SDF using user states and timers. cc: @robertwb

Would you like to explain more about "because keys are not guaranteed to be processed on the same operator instance." ? Is it because we check stateful DoFn first?

It's because we partition based on the element instead of on the key. This wouldn't work with stateful operations where we expect all keys to land on the same partition.

boyuanzz · 2020-10-27T22:26:06Z

Run Java Flink PortableValidatesRunner Batch

boyuanzz · 2020-10-27T22:26:18Z

Run Java Flink PortableValidatesRunner Streaming

boyuanzz · 2020-10-28T18:52:33Z

Run Java Flink PortableValidatesRunner Streaming

boyuanzz · 2020-10-29T03:19:51Z

...tion/src/main/java/org/apache/beam/runners/fnexecution/control/BundleCheckpointHandlers.java

+              outputTimestamp = Math.min(outputTimestamp, outputWatermark.getSeconds() * 1000);
+            }
+          } else {
+            outputTimestamp = BoundedWindow.TIMESTAMP_MIN_VALUE.getMillis();


Is it a good idea to rely on TimerInternals.getCurrentInputWatermark() to not hold back the watermark so much?
@mxm

You could use keyedStateInternals.minWatermarkHoldMs().

Thanks, Max! I just double checked the contract we have:

beam/model/fn-execution/src/main/proto/beam_fn_api.proto

Lines 236 to 243 in 7eb0e95

// The map is keyed by the local output name of the PTransform. Each

// value represents a lower bound on the timestamps of elements that

// are produced by this PTransform into each of its output PCollections

// when invoked with this application.

//

// If there is no watermark reported from RestrictionTracker, the runner will

// use MIN_TIMESTAMP by default.

map<string, google.protobuf.Timestamp> output_watermarks = 4;

Let's stick with MIN_TIMESTAMP for now.

boyuanzz · 2020-10-30T18:01:36Z

@mxm Would you like to take another pass on this PR?

mxm · 2020-10-30T18:48:20Z

...rg/apache/beam/runners/flink/translation/wrappers/streaming/ExecutableStageDoFnOperator.java

@@ -502,6 +690,8 @@ public void close() throws Exception {
    processWatermark1(Watermark.MAX_WATERMARK);
    while (getCurrentOutputWatermark() < Watermark.MAX_WATERMARK.getTimestamp()) {
      invokeFinishBundle();
+      // Sleep for 5s to wait for any timer to be fired.
+      Thread.sleep(5000);


This shouldn't be necessary. All timers will be drained on close(), at least in super.close().

From my testing, I found that if we don't free the occupation here, the DoFnOperator.onProcessingTime will never get a chance to be executed.

Kindly pinged : )

This is a bug then. All processing timers should be fired on close(). It would be great to get rid of the sleep which can have unexpected side effects, e.g. when operators are chained in Flink.

This was not a bug before we support sdf initiate checkpoint because normal processing time timer doesn't hold watermark. Now we are setting processing time timer to reschedule sdf residuals which holds the watermark back. When entering ExecutableStageDoFnOperator.close(), we will always in the while loop until all sdf processing timers get fired and watermark holds are removed. So the comment // Sleep for 5s to wait for any timer to be fired. is not precise. The comment should be // Sleep for 5s to wait for any SDF timer to be fired.

I found that Thread.sleep(5s) is not a correct solution. The reason why it worked occasionally is that the bundle is small enough and I got luck on race condition. I think we should figure out a way to be able to fire the SDF processing time timer when a bundle is closed within the life cycle of one ExecutableStageDoFnOperator.
Please correct me if I'm understanding it wrong:

Flink starts all operators at the same time and closes the operators when the input watermark reaches MAX_TIMESTAMP, or it closes operators in a reverse topological order and close() is a blocking call?

The processing time timers will not be fired anymore by the system once the operator.close() is invoked.

The assumption around ExecutableStageDoFnOperator is that there is only one bundle executing inside one operator. When the output watermark advances to MAX_TIMESTAMP, we consider this bundle completed.

With supporting SDF initiated checkpoint, we do need to have several bundles invoked inside one ExecutableStageDoFnOperator life cycle, which means we either:

Enable Flink to fire processing time timers after Operator.close() is invoked -- this may not be preferrable.

Or we try to close the bundle before we reach to the Operator.close().

Or we manually drain SDF timers with scarifying the ability of resumeDelay(). For example, the user may want to reschedule the SDF residuals in 5 mins but we have to fire it now.

Do you have any ideas/suggestions? Thanks for your help!

Double check the implementation of DoFnOperator and ExecutableStageDoFnOperator, we have already invoked finishBundle when reaching 1000 input elements or 1s processing time by default.
The real problem for SDF is that it's natural for SDF to read from Impluse and execute as a high fan-out DoFn. Based on current structure, once Impluse finishes, close() of SDF operator will be called, but meanwhile no more processing time timer can be registered. Simply draining timers from operator itself is not ideal.
Is it possible for us to change something here? For example, the operator should wait for global watermark advancing to MAX_TIMESTAMP to finish? Or the task should invokes operator.close() when global watermark advancing to MAX_TIMESTAMP?

Commit 56a004d introduces a workaround: draining processing timers manually for SDF. It's a workaround since it doesn't work efficiently if upstream advances watermark to MAX_TIMESTAMP very quickly. PTAL, @mxm

What you describe happens when you enable checkpointing or when you pass in --shutdownSourcesAfterIdleMs. The operator doesn't shutdown then and continues to process timers. The reason we do not enable this by default is that we want to cleanly shutdown pipelines during tests. Once an operator has been shut down, the entire job cannot be checkpointed anymore.

The new workaround looks good to me. We do the same in the super.close()method.

Thanks for the explanation! It seems like this problem has got attention from Flink: https://issues.apache.org/jira/browse/FLINK-18647.

I'm still feeling confused on when the operator.close() will be called. It seems like it happens under several condition:

When checkpointing happens

When source operator closes successfully, the next downstream operator's close() will be called.

Do I understand it correctly?

mxm · 2020-11-09T11:00:31Z

...rg/apache/beam/runners/flink/translation/wrappers/streaming/ExecutableStageDoFnOperator.java

@@ -502,6 +692,10 @@ public void close() throws Exception {
    processWatermark1(Watermark.MAX_WATERMARK);
    while (getCurrentOutputWatermark() < Watermark.MAX_WATERMARK.getTimestamp()) {
      invokeFinishBundle();
+      if (hasSdfProcessFn) {
+        // Sleep for 5s to wait for any SDF timer to be fired.
+        Thread.sleep(5000);


Perhaps add a TODO here, since we may want to eventually get rid of the sleep?

Thanks! Filed here: https://issues.apache.org/jira/browse/BEAM-11210

boyuanzz · 2020-11-09T20:27:50Z

Run Java Flink PortableValidatesRunner Batch

boyuanzz · 2020-11-09T20:28:01Z

Run Java Flink PortableValidatesRunner Streaming

boyuanzz · 2020-11-09T22:30:56Z

Run Java Flink PortableValidatesRunner Batch

boyuanzz · 2020-11-09T22:31:05Z

Run Java Flink PortableValidatesRunner Streaming

boyuanzz · 2020-11-11T01:31:28Z

Run Java Flink PortableValidatesRunner Batch

boyuanzz · 2020-11-11T01:31:42Z

Run Java Flink PortableValidatesRunner Streaming

boyuanzz · 2020-11-11T22:43:00Z

Python tests / Python Unit Tests (windows-latest, 3.6, py36) is not related.

I'm going to merge this PR. Thanks for your help!

mxm · 2020-11-12T09:22:33Z

My pleasure!

boyuanzz force-pushed the flink_sdf branch from cdf8629 to a1997cf Compare October 21, 2020 19:38

boyuanzz changed the title ~~[WIP] Initial support for self-checkpoint~~ [BEAM-10940] Add sdf initiated checkpoint support to portable Flink. Oct 21, 2020

boyuanzz force-pushed the flink_sdf branch from 9c751eb to 7eeb7f2 Compare October 23, 2020 00:09

boyuanzz force-pushed the flink_sdf branch from 7eeb7f2 to 5214f7f Compare October 23, 2020 02:14

boyuanzz force-pushed the flink_sdf branch from 7559089 to ce161e6 Compare October 23, 2020 20:00

boyuanzz force-pushed the flink_sdf branch from a293f01 to 82c2fc2 Compare October 26, 2020 22:02

mxm reviewed Oct 27, 2020

View reviewed changes

boyuanzz commented Oct 29, 2020

View reviewed changes

boyuanzz mentioned this pull request Oct 30, 2020

[BEAM-11070] Skip py bundle finalization test. #13228

Merged

4 tasks

mxm reviewed Oct 30, 2020

View reviewed changes

boyuanzz mentioned this pull request Nov 3, 2020

[BEAM-11187] Exclude SDF test suite because it requires support of self-checkpoint. #13250

Merged

4 tasks

mxm approved these changes Nov 9, 2020

View reviewed changes

boyuanzz force-pushed the flink_sdf branch from d8bb778 to 0ba5b74 Compare November 9, 2020 20:26

boyuanzz force-pushed the flink_sdf branch from 0ba5b74 to 1096d39 Compare November 9, 2020 22:00

Add sdf initiated checkpoint support to portable Flink.

28914c2

boyuanzz force-pushed the flink_sdf branch from 56a004d to 28914c2 Compare November 11, 2020 19:31

boyuanzz merged commit cf6fd1c into apache:master Nov 11, 2020

	// The map is keyed by the local output name of the PTransform. Each
	// value represents a lower bound on the timestamps of elements that
	// are produced by this PTransform into each of its output PCollections
	// when invoked with this application.
	//
	// If there is no watermark reported from RestrictionTracker, the runner will
	// use MIN_TIMESTAMP by default.
	map<string, google.protobuf.Timestamp> output_watermarks = 4;

[BEAM-10940] Add sdf initiated checkpoint support to portable Flink. #13105

[BEAM-10940] Add sdf initiated checkpoint support to portable Flink. #13105

Conversation

boyuanzz commented Oct 14, 2020 • edited

Post-Commit Tests Status (on master branch)

Pre-Commit Tests Status (on master branch)

GitHub Actions Tests Status (on master branch)

boyuanzz commented Oct 21, 2020

boyuanzz commented Oct 21, 2020

boyuanzz commented Oct 21, 2020

boyuanzz commented Oct 21, 2020

boyuanzz commented Oct 21, 2020

boyuanzz commented Oct 21, 2020

boyuanzz commented Oct 21, 2020

boyuanzz commented Oct 22, 2020

boyuanzz commented Oct 22, 2020

boyuanzz commented Oct 22, 2020

boyuanzz commented Oct 22, 2020 • edited

boyuanzz commented Oct 22, 2020

boyuanzz commented Oct 22, 2020

boyuanzz commented Oct 22, 2020

boyuanzz commented Oct 22, 2020

boyuanzz commented Oct 22, 2020

boyuanzz commented Oct 23, 2020

boyuanzz commented Oct 23, 2020

boyuanzz commented Oct 23, 2020

boyuanzz commented Oct 23, 2020

boyuanzz commented Oct 23, 2020

boyuanzz commented Oct 23, 2020

boyuanzz commented Oct 23, 2020

boyuanzz commented Oct 23, 2020

boyuanzz commented Oct 26, 2020

boyuanzz commented Oct 26, 2020

boyuanzz commented Oct 26, 2020

boyuanzz commented Oct 26, 2020

boyuanzz commented Oct 26, 2020

boyuanzz commented Oct 26, 2020

boyuanzz commented Oct 27, 2020

Choose a reason for hiding this comment

boyuanzz Oct 27, 2020 • edited

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

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

boyuanzz commented Oct 27, 2020

boyuanzz commented Oct 27, 2020

boyuanzz commented Oct 28, 2020

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

boyuanzz commented Oct 30, 2020

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

boyuanzz Nov 10, 2020 • edited

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

boyuanzz commented Nov 9, 2020

boyuanzz commented Nov 9, 2020

boyuanzz commented Oct 14, 2020 •

edited

boyuanzz commented Oct 22, 2020 •

edited

boyuanzz Oct 27, 2020 •

edited

boyuanzz Nov 10, 2020 •

edited