[SPARK-51714][SS] Add Failure Ingestion test to test state store checkpoint format V2 #50508

siying · 2025-04-03T18:46:37Z

Why are the changes needed?

The new state store checkpoint format needs failure tolerance tests to make sure the implementation is correct and delivers the behavior we would like.

Does this PR introduce any user-facing change?

No.

How was this patch tested?

It is test code itself

Was this patch authored or co-authored using generative AI tooling?

No.

…kpoint format V2

anishshri-db · 2025-04-03T18:52:53Z

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

@@ -135,7 +135,23 @@ class RocksDB(
  private val nativeStats = rocksDbOptions.statistics()

  private val workingDir = createTempDir("workingDir")
-  private val fileManager = new RocksDBFileManager(dfsRootDir, createTempDir("fileManager"),
+
+  protected def CreateFileManager(


nit: createFileManager

anishshri-db · 2025-04-03T18:53:17Z

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

@@ -132,8 +132,12 @@ class RocksDBFileManager(

  import RocksDBImmutableFile._

+  protected def GetFileSystem(myDfsRootDir: String, myHadoopConf: Configuration) : FileSystem = {


nit: getFileSystem

anishshri-db · 2025-04-03T18:53:34Z

...rc/main/scala/org/apache/spark/sql/execution/streaming/state/RocksDBStateStoreProvider.scala

@@ -524,13 +524,33 @@ private[sql] class RocksDBStateStoreProvider
  @volatile private var stateStoreEncoding: String = _
  @volatile private var stateSchemaProvider: Option[StateSchemaProvider] = _

+  protected def CreateRocksDB(


Could we move all these functions to start with lower case ?

anishshri-db · 2025-04-03T18:54:07Z

...a/org/apache/spark/sql/execution/streaming/state/FailureInjectionCheckpointFileManager.scala

+import org.apache.spark.sql.execution.streaming.CheckpointFileManager.{CancellableFSDataOutputStream, RenameBasedFSDataOutputStream}
+import org.apache.spark.sql.execution.streaming.FileSystemBasedCheckpointFileManager
+
+/** A wrapper file output stream that will throw exception in close() and put the underlying


nit: lets follow the comment style ?

anishshri-db · 2025-04-03T20:24:16Z

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

+    stopMaintenanceTaskNoLock()
+  }
+
+  /** Only used for unit tests. The function doesn't hold lockloadedProviders. Calling


nit: doesn't hold the loadedProviders lock

anishshri-db · 2025-04-03T20:24:34Z

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

+
+  /** Only used for unit tests. The function doesn't hold lockloadedProviders. Calling
+   * it can work-around a deadlock condition where a maintenance task is waiting for the lock */
+  private[streaming] def stopMaintenanceTaskNoLock(): Unit = {


nit: stopMaintenanceTaskWithoutLock

anishshri-db · 2025-04-03T20:25:40Z

...a/org/apache/spark/sql/execution/streaming/state/FailureInjectionCheckpointFileManager.scala

+  override def close(): Unit = {
+    if (!closed) {
+      closed = true
+      FailureInjectionFileSystem.delayedStreams =


we are adding to a singleton here ?

Yes, it is a singleton.

Updating singleton fields within this class feels a bit odd. Do you think we can refactor a bit ?

@anishshri-db it's very hard to access any of these objects from the test code, as they are created deep in the state store code as we are operating on the DB level. Static variables are the best way of communicating between unit test and file system wrapper in deep level that I can think of. What's the main concern of using singleton here?

Hmm ok - wonder if there is some way to be keep this test/class local. but if not, its fine. also cc - @HeartSaVioR - in case he has seen other patterns for similar cases before

If the singleton is not reused among suites, it's probably fine.

But I'm concerned the status of object instance are "shared" and there is no reset to make sure the test is isolated.

Shall we have a clear() method to reset the status, and ensure tests relying on this object to call that method in prior?

Also, I'd feel more comfort to move the content of this file to be in RocksDBCheckpointFailureInjectionSuite, since the object instance is "unsafe" to be used among test suites. (There are some environments where multiple test suites can be executed in parallel.)

anishshri-db · 2025-04-03T20:25:59Z

...a/org/apache/spark/sql/execution/streaming/state/FailureInjectionCheckpointFileManager.scala

+  }
+}
+
+/** Contains a list of variables for failure ingestion conditions */


nit: could we add some more details here ?

anishshri-db · 2025-04-03T20:26:14Z

...a/org/apache/spark/sql/execution/streaming/state/FailureInjectionCheckpointFileManager.scala

+  override def open(f: Path, bufferSize: Int): FSDataInputStream = innerFs.open(f, bufferSize)
+
+  override def create(
+                       f: Path,


nit: indent seems off ?

anishshri-db · 2025-04-03T20:26:33Z

...a/org/apache/spark/sql/execution/streaming/state/FailureInjectionCheckpointFileManager.scala

+  override def delete(f: Path, recursive: Boolean): Boolean = innerFs.delete(f, recursive)
+
+  override def listStatus(f: Path): Array[FileStatus] = {
+    innerFs.listStatus(f)


could move to same line to be consistent ?

anishshri-db · 2025-04-03T20:26:45Z

...a/org/apache/spark/sql/execution/streaming/state/FailureInjectionCheckpointFileManager.scala

+/** A rapper RocksDB State Store Provider that replaces FileSystem used in RocksDBFileManager
+ * to FailureInjectionFileSystem. */
+class FailureInjectionRocksDBStateStoreProvider extends RocksDBStateStoreProvider {
+  override def CreateRocksDB(


nit: createRocksDB

anishshri-db · 2025-04-03T20:26:57Z

...a/org/apache/spark/sql/execution/streaming/state/FailureInjectionCheckpointFileManager.scala

+  /** RocksDBFieManager is created by RocksDB class where it creates a default FileSystem.
+   * we made RocksDB create a RocksDBFileManager but a different FileSystem here. */
+  def createRocksDBWithFaultInjection(
+                                       dfsRootDir: String,


nit: indent seems off ?

anishshri-db · 2025-04-03T20:50:09Z

...a/org/apache/spark/sql/execution/streaming/state/FailureInjectionCheckpointFileManager.scala

+    } else {
+      ret
+    }
+


nit: extra newline ?

anishshri-db · 2025-04-03T20:50:19Z

...a/org/apache/spark/sql/execution/streaming/state/FailureInjectionCheckpointFileManager.scala

+
+  override def renameTempFile(srcPath: Path, dstPath: Path,
+                              overwriteIfPossible: Boolean): Unit = {
+    if (FailureInjectionFileSystem.allowOverwriteInRename || !fs.exists(dstPath)) {


nit: indent seems off ?

anishshri-db · 2025-04-03T20:51:02Z

...a/org/apache/spark/sql/execution/streaming/state/FailureInjectionCheckpointFileManager.scala

+      partitionId = partitionId
+    ) {
+      override def createFileManager(
+                                      dfsRootDir: String,


nit: indent

anishshri-db · 2025-04-03T20:51:07Z

...a/org/apache/spark/sql/execution/streaming/state/FailureInjectionCheckpointFileManager.scala

+        ) {
+          override def getFileSystem(
+                                      myDfsRootDir: String,
+                                      myHadoopConf: Configuration): FileSystem = {


nit: indent

anishshri-db · 2025-04-05T00:25:36Z

...a/org/apache/spark/sql/execution/streaming/state/FailureInjectionCheckpointFileManager.scala

+object FailureInjectionRocksDBStateStoreProvider {
+  /**
+   * RocksDBFieManager is created by RocksDB class where it creates a default FileSystem.
+   * we made RocksDB create a RocksDBFileManager but a different FileSystem here.


nit: We make RocksDB create a RocksDBFileManager that uses a different FileSystem here

anishshri-db · 2025-04-05T00:30:47Z

.../org/apache/spark/sql/execution/streaming/state/RocksDBCheckpointFailureInjectionSuite.scala

+
+@SlowSQLTest
+/**
+ * Test suite to ingest some failures in RocksDB checkpoint */


nit: inject some failures ?

anishshri-db · 2025-04-05T00:32:58Z

.../org/apache/spark/sql/execution/streaming/state/RocksDBCheckpointFailureInjectionSuite.scala

+   *  be killed or write zip file. Only after the later one is successfully committed, it comes back
+   *  and write the zip file.
+   *  */
+  test("Zip File Overwritten by Previous Task Checkpoint V2") {


Is this largely the same as prev test - only for checkpoint v2 ? should we parameterize into 1 test ?

anishshri-db · 2025-04-05T00:33:12Z

.../org/apache/spark/sql/execution/streaming/state/RocksDBCheckpointFailureInjectionSuite.scala

+   * be killed or write changelog file. Only after the later one is successfully committed, it come
+   * back and write the changelog file.
+   *  */
+  test("Changelog File Overwritten by Previous Task With Changelog Checkpoint V2") {


Same here ?

anishshri-db · 2025-04-05T00:33:37Z

.../org/apache/spark/sql/execution/streaming/state/RocksDBCheckpointFailureInjectionSuite.scala

+   * 2. The batch eventually failed
+   * 3. Query is retried and moved forward
+   * 4. The snapshot checkpoint succeeded
+   * In checkpoint V2, this snapshot shouldn't take effective. Otherwise, it will break the strong


nit: shouldn't take effect

anishshri-db · 2025-04-05T00:35:21Z

.../org/apache/spark/sql/execution/streaming/state/RocksDBCheckpointFailureInjectionSuite.scala

+    }
+  }
+
+  test("Basic RocksDB Zip File Upload Failure Handling") {


Dont we need to run test with both checkpoint v1 and v2 ?

anishshri-db · 2025-04-05T22:12:12Z

.../org/apache/spark/sql/execution/streaming/state/RocksDBCheckpointFailureInjectionSuite.scala

+
+  /**
+   * This test is to simulate the case where a previous task had connectivity problem that couldn't
+   * be killed or write changelog file. Only after the later one is successfully committed, it come


nit: comes back and writes the

anishshri-db · 2025-04-05T22:13:29Z

.../org/apache/spark/sql/execution/streaming/state/RocksDBCheckpointFailureInjectionSuite.scala

+            db.put("foo", "bar")
+            checkpointId3 = commitAndGetCheckpointId(db)
+
+            db2.doMaintenance()


Do we have a test with varying maintenance intervals ? i.e. where maintenance is configured to run very often vs very rarely ? maybe not for each case - but for some of the most common test cases here ?

We don't have it. Maybe do it as a follow up when we have time later.

anishshri-db · 2025-04-05T22:13:43Z

.../org/apache/spark/sql/execution/streaming/state/RocksDBCheckpointFailureInjectionSuite.scala

+  }
+
+  /**
+   * An integreated test to cover this scenario:


nit: An integrated test

anishshri-db · 2025-04-05T22:17:23Z

...a/org/apache/spark/sql/execution/streaming/state/FailureInjectionCheckpointFileManager.scala

+}
+
+/**
+ * A wrapper checkpoint file manager that might inject functions in some function calls.


nit: inject failures in some function calls ?

anishshri-db · 2025-04-05T22:17:51Z

...a/org/apache/spark/sql/execution/streaming/state/FailureInjectionCheckpointFileManager.scala

+ * This can be put into SQLConf.STREAMING_CHECKPOINT_FILE_MANAGER_CLASS to provide failure
+ * injection behavior.
+ *
+ * @param path


nit: you need to add comments for the args ?

anishshri-db

lgtm pending comments being addressed. Thx

siying · 2025-04-06T03:38:50Z

Thanks @anishshri-db for review and approval. @HeartSaVioR can you help review and merge it?

HeartSaVioR

Will review from L385 of the last test suite file.

HeartSaVioR · 2025-04-08T06:57:52Z

...a/org/apache/spark/sql/execution/streaming/state/FailureInjectionCheckpointFileManager.scala

+  override def close(): Unit = {
+    if (!closed) {
+      closed = true
+      FailureInjectionFileSystem.delayedStreams =


But I'm concerned the status of object instance are "shared" and there is no reset to make sure the test is isolated.

Shall we have a clear() method to reset the status, and ensure tests relying on this object to call that method in prior?

Also, I'd feel more comfort to move the content of this file to be in RocksDBCheckpointFailureInjectionSuite, since the object instance is "unsafe" to be used among test suites. (There are some environments where multiple test suites can be executed in parallel.)

HeartSaVioR · 2025-04-08T07:09:34Z

...a/org/apache/spark/sql/execution/streaming/state/FailureInjectionCheckpointFileManager.scala

+   * RocksDBFieManager is created by RocksDB class where it creates a default FileSystem.
+   * We make RocksDB create a RocksDBFileManager that uses a different FileSystem here.
+   * */
+  def createRocksDBWithFaultInjection(


nit: any reason this has to be placed in different object / method? If this is only used once in createRocksDB, I'm not sure this is quite different from inlining.

This is actually used twice. One in FailureInjectionRocksDBStateStoreProvider.createRocksDB() and one in withDBT

HeartSaVioR · 2025-04-08T07:11:52Z

...a/org/apache/spark/sql/execution/streaming/state/FailureInjectionCheckpointFileManager.scala

+object FailureInjectionFileSystem {
+  // File names matching this regex will cause the copyFromLocalFile to fail
+  var failPreCopyFromLocalFileNameRegex: Seq[String] = Seq.empty
+  // File names matching this regex will cause the createAtomic to fail and put the streams in


nit: is it close() rather than createAtomic()?

@siying Could you please revisit this?

HeartSaVioR · 2025-04-08T07:12:55Z

.../org/apache/spark/sql/execution/streaming/state/RocksDBCheckpointFailureInjectionSuite.scala

+
+@SlowSQLTest
+/**
+ * Test suite to inject some failures in RocksDB checkpoint */


nit: please use /** ... */ in the same line for one-liner classdoc.

HeartSaVioR · 2025-04-08T07:13:45Z

.../org/apache/spark/sql/execution/streaming/state/RocksDBCheckpointFailureInjectionSuite.scala

+
+  override def beforeEach(): Unit = {
+    super.beforeEach()
+    FailureInjectionFileSystem.failPreCopyFromLocalFileNameRegex = Seq.empty


nit: likewise I said, shall we just have a method to clear everything instead of doing here?

HeartSaVioR · 2025-04-08T07:29:30Z

.../org/apache/spark/sql/execution/streaming/state/RocksDBCheckpointFailureInjectionSuite.scala

+                ex,
+                condition = "CANNOT_LOAD_STATE_STORE.CANNOT_READ_STREAMING_STATE_FILE",
+                parameters = Map(
+                  "fileToRead" -> s"$remoteDir/2.changelog"


Wait... didn't we disable changelog checkpointing?

Yes, but I think that's always the error we throw, even if changelog is disabled.

OK that is odd. @anishshri-db Would you mind taking a look?

HeartSaVioR · 2025-04-08T07:30:57Z

.../org/apache/spark/sql/execution/streaming/state/RocksDBCheckpointFailureInjectionSuite.scala

+    }
+  }
+
+  Seq(false, true).foreach { ifEnableStateStoreCheckpointIds =>


This is too similar with the above test except the pattern of file we want to block uploading. Shall we parameterize and dedup the code?

HeartSaVioR · 2025-04-08T07:31:24Z

.../org/apache/spark/sql/execution/streaming/state/RocksDBCheckpointFailureInjectionSuite.scala

+    test(
+      "Zip File Overwritten by Previous Task Checkpoint " +
+      s"ifEnableStateStoreCheckpointIds = $ifEnableStateStoreCheckpointIds") {
+      val fmClass = "org.apache.spark.sql.execution.streaming.state." +


HeartSaVioR · 2025-04-08T07:41:49Z

.../org/apache/spark/sql/execution/streaming/state/RocksDBCheckpointFailureInjectionSuite.scala

+              // This validation isn't necessary here but we just would like to make sure
+              // FailureInjectionCheckpointFileManager has correct behavior --  allows zip files
+              // to be delayed to be written, so that the test for
+              // ifEnableStateStoreCheckpointIds = trrue is valid.


nit: not trrue

HeartSaVioR · 2025-04-08T08:14:59Z

.../org/apache/spark/sql/execution/streaming/state/RocksDBCheckpointFailureInjectionSuite.scala

+   * consistency guaranteed by V2.
+   */
+  test("Delay Snapshot V2") {
+    val fmClass = "org.apache.spark.sql.execution.streaming.state." +


HeartSaVioR

First pass

HeartSaVioR · 2025-04-08T12:40:44Z

.../org/apache/spark/sql/execution/streaming/state/RocksDBCheckpointFailureInjectionSuite.scala

+   */
+  Seq(false, true).foreach { ifAllowRenameOverwrite =>
+    test(s"Job failure with changelog shows up ifAllowRenameOverwrite = $ifAllowRenameOverwrite") {
+      val fmClass = "org.apache.spark.sql.execution.streaming.state." +


HeartSaVioR · 2025-04-08T12:42:39Z

.../org/apache/spark/sql/execution/streaming/state/RocksDBCheckpointFailureInjectionSuite.scala

+              additionalConfs = Map(
+                rocksdbChangelogCheckpointingConfKey -> "true",
+                SQLConf.STATE_STORE_CHECKPOINT_FORMAT_VERSION.key -> "2",
+                STREAMING_CHECKPOINT_FILE_MANAGER_CLASS.parent.key -> fmClass)),


What is parent here? (I haven't realized that it had been referred as parent for other occurrences).

I copied it from other test suites. tests such as CheckpointFileManagerSuite, RocksDBSuite and other several tests are all doing this.

HeartSaVioR · 2025-04-08T12:43:31Z

.../org/apache/spark/sql/execution/streaming/state/RocksDBCheckpointFailureInjectionSuite.scala

+            ExpectFailure[SparkException] { ex =>
+              ex.getCause.getMessage.contains("CANNOT_WRITE_STATE_STORE.CANNOT_COMMIT")
+            },
+            AddData(inputData, 3, 1)


Is it intentional to add more data after the query is expected to fail? Because it could be just confusing and you can just do inputData.addData() outside of testStream.

HeartSaVioR · 2025-04-08T12:46:07Z

.../org/apache/spark/sql/execution/streaming/state/RocksDBCheckpointFailureInjectionSuite.scala

+   * and validate that the snapshot checkpoint is not used in subsequent query restart.
+   */
+  test("Previous Maintenance Snapshot Checkpoint Overwrite") {
+    val fmClass = "org.apache.spark.sql.execution.streaming.state." +


HeartSaVioR · 2025-04-08T12:49:27Z

.../org/apache/spark/sql/execution/streaming/state/RocksDBCheckpointFailureInjectionSuite.scala

+          CheckAnswer((3, 1)),
+          AddData(inputData, 3, 2),
+          AddData(inputData, 3, 1),
+          CheckAnswer((3, 1), (3, 3), (2, 1), (1, 1)),


nit: I'd rather use CheckNewAnswer to avoid accumulation of results.

siying

Addressed most comments.

siying · 2025-04-08T20:58:08Z

...a/org/apache/spark/sql/execution/streaming/state/FailureInjectionCheckpointFileManager.scala

+   * RocksDBFieManager is created by RocksDB class where it creates a default FileSystem.
+   * We make RocksDB create a RocksDBFileManager that uses a different FileSystem here.
+   * */
+  def createRocksDBWithFaultInjection(


This is actually used twice. One in FailureInjectionRocksDBStateStoreProvider.createRocksDB() and one in withDBT

siying · 2025-04-08T21:03:47Z

.../org/apache/spark/sql/execution/streaming/state/RocksDBCheckpointFailureInjectionSuite.scala

+                ex,
+                condition = "CANNOT_LOAD_STATE_STORE.CANNOT_READ_STREAMING_STATE_FILE",
+                parameters = Map(
+                  "fileToRead" -> s"$remoteDir/2.changelog"


Yes, but I think that's always the error we throw, even if changelog is disabled.

siying · 2025-04-08T22:02:51Z

.../org/apache/spark/sql/execution/streaming/state/RocksDBCheckpointFailureInjectionSuite.scala

+              additionalConfs = Map(
+                rocksdbChangelogCheckpointingConfKey -> "true",
+                SQLConf.STATE_STORE_CHECKPOINT_FORMAT_VERSION.key -> "2",
+                STREAMING_CHECKPOINT_FILE_MANAGER_CLASS.parent.key -> fmClass)),


I copied it from other test suites. tests such as CheckpointFileManagerSuite, RocksDBSuite and other several tests are all doing this.

HeartSaVioR

Only nit here.

https://github.com/apache/spark/pull/50508/files/f58f00b2d20512e007f1c8f9b3ee4dea626cbeeb#r2034392980

Thanks for the work!

...a/org/apache/spark/sql/execution/streaming/state/FailureInjectionCheckpointFileManager.scala

…ng/state/FailureInjectionCheckpointFileManager.scala review comment

HeartSaVioR · 2025-04-09T04:47:19Z

507c30c

This passed CI, and my new commit does not change anything except code comment.

eb3ce1e

Hence I'll consider this as CI passed.

HeartSaVioR

+1

HeartSaVioR · 2025-04-09T04:48:53Z

Thanks! Merging to master ~~/4.0 (this wouldn't be risky since it's test only change)~~.

The main change is the test, but we also change the main code, so I can't say this is not risky.

siying · 2025-04-09T18:10:54Z

@HeartSaVioR thank you so much for your review and modification before pushing.

github-actions bot added SQL STRUCTURED STREAMING labels Apr 3, 2025

[SPARK-51714][SS] Add Failure Ingestion test to test state store chec…

a4656e8

…kpoint format V2

siying force-pushed the ingest_failure9 branch from 28fb3d8 to a4656e8 Compare April 3, 2025 18:48

anishshri-db reviewed Apr 3, 2025

View reviewed changes

siying added 3 commits April 3, 2025 20:27

Address comments

2aba4b9

Address comments

595b09d

minor style

c3c1857

anishshri-db reviewed Apr 3, 2025

View reviewed changes

siying added 3 commits April 3, 2025 20:57

comments and fix

83bd5be

fix

a051c7d

fix

edd7ef6

anishshri-db reviewed Apr 5, 2025

View reviewed changes

Address comments.

4e9b363

anishshri-db reviewed Apr 5, 2025

View reviewed changes

anishshri-db approved these changes Apr 5, 2025

View reviewed changes

Address comments

72bbd5b

HeartSaVioR reviewed Apr 8, 2025

View reviewed changes

Address comments

f58f00b

siying commented Apr 8, 2025

View reviewed changes

Add volatile

507c30c

HeartSaVioR reviewed Apr 9, 2025

View reviewed changes

...a/org/apache/spark/sql/execution/streaming/state/FailureInjectionCheckpointFileManager.scala Outdated Show resolved Hide resolved

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

eb3ce1e

…ng/state/FailureInjectionCheckpointFileManager.scala review comment

HeartSaVioR approved these changes Apr 9, 2025

View reviewed changes

HeartSaVioR changed the title ~~[SPARK-51714][SS] Add Failure Ingestion test to test state store checkpoint format V2~~ [SPARK-51714][SS][TESTS] Add Failure Ingestion test to test state store checkpoint format V2 Apr 9, 2025

HeartSaVioR changed the title ~~[SPARK-51714][SS][TESTS] Add Failure Ingestion test to test state store checkpoint format V2~~ [SPARK-51714][SS] Add Failure Ingestion test to test state store checkpoint format V2 Apr 9, 2025

HeartSaVioR closed this in c7024d3 Apr 9, 2025

		@@ -132,8 +132,12 @@ class RocksDBFileManager(

		import RocksDBImmutableFile._

		protected def GetFileSystem(myDfsRootDir: String, myHadoopConf: Configuration) : FileSystem = {

[SPARK-51714][SS] Add Failure Ingestion test to test state store checkpoint format V2 #50508

[SPARK-51714][SS] Add Failure Ingestion test to test state store checkpoint format V2 #50508

Conversation

siying commented Apr 3, 2025

Why are the changes needed?

Does this PR introduce any user-facing change?

How was this patch tested?

Was this patch authored or co-authored using generative AI tooling?

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

siying Apr 3, 2025 • 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

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

anishshri-db Apr 5, 2025 • edited Loading

Choose a reason for hiding this comment

anishshri-db left a comment

Choose a reason for hiding this comment

siying commented Apr 6, 2025

HeartSaVioR left a 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

HeartSaVioR left a 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

siying left a 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

HeartSaVioR left a comment

Choose a reason for hiding this comment

HeartSaVioR commented Apr 9, 2025

HeartSaVioR left a comment

siying Apr 3, 2025 •

edited

Loading

anishshri-db Apr 5, 2025 •

edited

Loading

HeartSaVioR commented Apr 9, 2025 •

edited

Loading