WweiL commented on code in PR #48460:
URL: https://github.com/apache/spark/pull/48460#discussion_r1896257961
##########
sql/core/src/test/scala/org/apache/spark/sql/execution/streaming/state/RocksDBStateStoreCheckpointFormatV2Suite.scala:
##########
@@ -211,6 +218,317 @@ class RocksDBStateStoreCheckpointFormatV2Suite extends
StreamTest
}
}
+ val changelogEnabled =
+ "spark.sql.streaming.stateStore.rocksdb.changelogCheckpointing.enabled" ->
"true"
+ val changelogDisabled =
+ "spark.sql.streaming.stateStore.rocksdb.changelogCheckpointing.enabled" ->
"false"
+ val ckptv2 = SQLConf.STATE_STORE_CHECKPOINT_FORMAT_VERSION.key -> "2"
+
+ case class checkpointV2CompatibilityTest(
+ configsForFirstRun: Seq[(String, String)],
+ configsForSecondRun: Seq[(String, String)]
+ ) {
+ override def toString: String = "first run - " +
+ s"changeLogEnabled: ${configsForFirstRun(0)._2}, ckpt ver:
${configsForFirstRun(1)._2}" +
+ s" - second run - changeLogEnabled: ${configsForSecondRun(0)._2}, " +
+ s"ckpt ver: ${configsForSecondRun(1)._2}"
+ }
+
+ val testConfigSetups = Seq(
+ // Enable and disable changelog under ckpt v2
+ checkpointV2CompatibilityTest(Seq(changelogEnabled, ckptv2),
Seq(changelogEnabled, ckptv2)),
+ checkpointV2CompatibilityTest(Seq(changelogDisabled, ckptv2),
Seq(changelogDisabled, ckptv2))
+ )
+
+ testConfigSetups.foreach { testConfig =>
+ testWithRocksDBStateStore("checkpointFormatVersion2 Backward Compatibility
- simple agg - " +
Review Comment:
Removed the tests
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
