eason-yuchen-liu commented on code in PR #46944:
URL: https://github.com/apache/spark/pull/46944#discussion_r1638672073
##########
sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/v2/state/StateDataSourceReadSuite.scala:
##########
@@ -194,6 +195,79 @@ class StateDataSourceNegativeTestSuite extends
StateDataSourceTestBase {
}
}
}
+
+ test("ERROR: snapshotStartBatchId specified to negative") {
+ withTempDir { tempDir =>
+ val exc = intercept[StateDataSourceInvalidOptionValueIsNegative] {
+ spark.read.format("statestore")
+ // trick to bypass getting the last committed batch before
validating operator ID
+ .option(StateSourceOptions.BATCH_ID, 0)
+ .option(StateSourceOptions.SNAPSHOT_START_BATCH_ID, -1)
+ .load(tempDir.getAbsolutePath)
+ }
+ checkError(exc, "STDS_INVALID_OPTION_VALUE.IS_NEGATIVE", "42616",
+ Map("optionName" -> StateSourceOptions.SNAPSHOT_START_BATCH_ID))
+ }
+ }
+
+ test("ERROR: snapshotPartitionId specified to negative") {
+ withTempDir { tempDir =>
+ val exc = intercept[StateDataSourceInvalidOptionValueIsNegative] {
+ spark.read.format("statestore")
+ // trick to bypass getting the last committed batch before
validating operator ID
+ .option(StateSourceOptions.BATCH_ID, 0)
+ .option(StateSourceOptions.SNAPSHOT_PARTITION_ID, -1)
+ .load(tempDir.getAbsolutePath)
+ }
+ checkError(exc, "STDS_INVALID_OPTION_VALUE.IS_NEGATIVE", "42616",
+ Map("optionName" -> StateSourceOptions.SNAPSHOT_PARTITION_ID))
+ }
+ }
+
+ test("ERROR: snapshotStartBatchId specified without snapshotPartitionId or
vice versa") {
+ withTempDir { tempDir =>
+ val exc = intercept[StateDataSourceUnspecifiedRequiredOption] {
+ spark.read.format("statestore")
+ // trick to bypass getting the last committed batch before
validating operator ID
+ .option(StateSourceOptions.BATCH_ID, 0)
+ .option(StateSourceOptions.SNAPSHOT_START_BATCH_ID, 0)
+ .load(tempDir.getAbsolutePath)
+ }
+ checkError(exc, "STDS_REQUIRED_OPTION_UNSPECIFIED", "42601",
+ Map("optionName" -> StateSourceOptions.SNAPSHOT_PARTITION_ID))
+ }
+
+ withTempDir { tempDir =>
+ val exc = intercept[StateDataSourceUnspecifiedRequiredOption] {
+ spark.read.format("statestore")
+ // trick to bypass getting the last committed batch before
validating operator ID
+ .option(StateSourceOptions.BATCH_ID, 0)
+ .option(StateSourceOptions.SNAPSHOT_PARTITION_ID, 0)
+ .load(tempDir.getAbsolutePath)
+ }
+ checkError(exc, "STDS_REQUIRED_OPTION_UNSPECIFIED", "42601",
+ Map("optionName" -> StateSourceOptions.SNAPSHOT_START_BATCH_ID))
+ }
+ }
+
+ test("ERROR: snapshotStartBatchId is greater than snapshotEndBatchId") {
+ withTempDir { tempDir =>
+ val startBatchId = 1
+ val endBatchId = 0
+ val exc = intercept[StateDataSourceInvalidOptionValue] {
+ spark.read.format("statestore")
+ // trick to bypass getting the last committed batch before
validating operator ID
+ .option(StateSourceOptions.BATCH_ID, 0)
Review Comment:
Do you mean the comment or the code? The code cannot be removed, otherwise
it will try to look for the max value of batch id available and an error will
throw since nothing is there. Above tests use this trick a lot.
--
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]
