You are viewing a plain text version of this content. The canonical link for it is here.
Posted to reviews@spark.apache.org by "zifeif2 (via GitHub)" <gi...@apache.org> on 2026/01/08 00:29:28 UTC
[PR] [SPARK-54590][SS] Support CheckpointV2 for StatePartitionAllColumnFamiliesWriter [spark]
zifeif2 opened a new pull request, #53720:
URL: https://github.com/apache/spark/pull/53720
<!--
Thanks for sending a pull request! Here are some tips for you:
1. If this is your first time, please read our contributor guidelines: https://spark.apache.org/contributing.html
2. Ensure you have added or run the appropriate tests for your PR: https://spark.apache.org/developer-tools.html
3. If the PR is unfinished, add '[WIP]' in your PR title, e.g., '[WIP][SPARK-XXXX] Your PR title ...'.
4. Be sure to keep the PR description updated to reflect all changes.
5. Please write your PR title to summarize what this PR proposes.
6. If possible, provide a concise example to reproduce the issue for a faster review.
7. If you want to add a new configuration, please read the guideline first for naming configurations in
'core/src/main/scala/org/apache/spark/internal/config/ConfigEntry.scala'.
8. If you want to add or modify an error type or message, please read the guideline first in
'common/utils/src/main/resources/error/README.md'.
-->
### What changes were proposed in this pull request?
<!--
Please clarify what changes you are proposing. The purpose of this section is to outline the changes and how this PR fixes the issue.
If possible, please consider writing useful notes for better and faster reviews in your PR. See the examples below.
1. If you refactor some codes with changing classes, showing the class hierarchy will help reviewers.
2. If you fix some SQL features, you can provide some references of other DBMSes.
3. If there is design documentation, please add the link.
4. If there is a discussion in the mailing list, please add the link.
-->
Support checkpointV2 for repartition writer and StateRewriter by returning the checkpoint Id to caller function after write is done.
Changes include
- RocksDB loadWithCheckpointId supports loadEmpty
- StatePartitionAllColumnFamiliesWriter return StateStoreCheckpointInfo
- StateRewriter also propagate StateStoreCheckpointInfo back to the caller
### Why are the changes needed?
<!--
Please clarify why the changes are needed. For instance,
1. If you propose a new API, clarify the use case for a new API.
2. If you fix a bug, you can clarify why it is a bug.
-->
This is required in PrPr for repartition project
### Does this PR introduce _any_ user-facing change?
<!--
Note that it means *any* user-facing change including all aspects such as new features, bug fixes, or other behavior changes. Documentation-only updates are not considered user-facing changes.
If yes, please clarify the previous behavior and the change this PR proposes - provide the console output, description and/or an example to show the behavior difference if possible.
If possible, please also clarify if this is a user-facing change compared to the released Spark versions or within the unreleased branches such as master.
If no, write 'No'.
-->
No
### How was this patch tested?
<!--
If tests were added, say they were added here. Please make sure to add some test cases that check the changes thoroughly including negative and positive cases if possible.
If it was tested in a way different from regular unit tests, please clarify how you tested step by step, ideally copy and paste-able, so that other reviewers can test and check, and descendants can verify in the future.
If tests were not added, please describe why they were not added and/or why it was difficult to add.
If benchmark tests were added, please run the benchmarks in GitHub Actions for the consistent environment, and the instructions could accord to: https://spark.apache.org/developer-tools.html#github-workflow-benchmarks.
-->
See added unit tests on moth operator with single state store and multiple state stores
### Was this patch authored or co-authored using generative AI tooling?
<!--
If generative AI tooling has been used in the process of authoring this patch, please include the
phrase: 'Generated-by: ' followed by the name of the tool and its version.
If no, write 'No'.
Please refer to the [ASF Generative Tooling Guidance](https://www.apache.org/legal/generative-tooling.html) for details.
-->
Yes. Sonnet 4.5
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-54590][SS] Support CheckpointV2 for StatePartitionAllColumnFamiliesWriter [spark]
Posted by "zifeif2 (via GitHub)" <gi...@apache.org>.
zifeif2 commented on code in PR #53720:
URL: https://github.com/apache/spark/pull/53720#discussion_r2691650963
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/state/StateRewriter.scala:
##########
@@ -342,6 +381,25 @@ class StateRewriter(
None
}
}
+
+ private def setCorrectCheckpointVersion(): Unit = {
+ // Setting checkpoint version in sqlConf based on previous commitLog in case user forgot to
+ // set STATE_STORE_CHECKPOINT_FORMAT_VERSION and crash rewriter.
+ // Using read batch commit since the latest commit could be a skipped batch
+ // If SQLConf.STATE_STORE_CHECKPOINT_FORMAT_VERSION is wrong,
+ // readCheckpoint.commitLog will throw an exception.
+ // This prevents the StateRewriter from failing to write the correct state files
+ try {
+ readCheckpoint.commitLog.get(readBatchId)
+ } catch {
+ case e: IllegalStateException =>
+ val sparkThrowable = e.getCause.asInstanceOf[SparkThrowable]
+ if (sparkThrowable.getCondition == "INVALID_LOG_VERSION.EXACT_MATCH_VERSION") {
+ sparkSession.conf.set(SQLConf.STATE_STORE_CHECKPOINT_FORMAT_VERSION.key,
+ sparkThrowable.getMessageParameters.get("version"))
Review Comment:
Didn't really like how this is implemented, but this is the only way I can think of. `readCheckpoint` is created with the sparkSession with the wrong `STATE_STORE_CHECKPOINT_FORMAT_VERSION` set (see line 85), so when we access commitLog, it will throw an error `INVALID_LOG_VERSION.EXACT_MATCH_VERSION` so we can't check if there is checkpointId in commitLog, but instead of just relying on the error message to set the correct checkpoint version.
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-54590][SS] Support Checkpoint V2 for State Rewriter and Repartitioning [spark]
Posted by "anishshri-db (via GitHub)" <gi...@apache.org>.
anishshri-db closed pull request #53720: [SPARK-54590][SS] Support Checkpoint V2 for State Rewriter and Repartitioning
URL: https://github.com/apache/spark/pull/53720
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-54590][SS] Support Checkpoint V2 for State Rewriter and Repartitioning [spark]
Posted by "zifeif2 (via GitHub)" <gi...@apache.org>.
zifeif2 commented on code in PR #53720:
URL: https://github.com/apache/spark/pull/53720#discussion_r2696211902
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/state/RocksDBFileManager.scala:
##########
@@ -375,6 +375,8 @@ class RocksDBFileManager(
createDfsRootDirIfNotExist()
// Since we cleared the local dir, we should also clear the local file mapping
rocksDBFileMapping.clear()
+ // Set empty metrics since we're not loading any files from DFS
+ loadCheckpointMetrics = RocksDBFileManagerMetrics.EMPTY_METRICS
Review Comment:
Added this line to make sure that loadCheckpointMetrics is set correctly when we are loading empty store cc @anishshri-db
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-54590][SS] Support CheckpointV2 for StatePartitionAllColumnFamiliesWriter [spark]
Posted by "zifeif2 (via GitHub)" <gi...@apache.org>.
zifeif2 commented on code in PR #53720:
URL: https://github.com/apache/spark/pull/53720#discussion_r2692246949
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/state/StateRewriter.scala:
##########
@@ -81,24 +82,46 @@ class StateRewriter(
private val stateRootLocation = new Path(
resolvedCheckpointLocation, StreamingCheckpointConstants.DIR_NAME_STATE).toString
- def run(): Unit = {
+ // return a Map[operator id, Array[partition -> Array[stateStore -> StateStoreCheckpointInfo]]]
+ def run(): Map[Long, Array[Array[StateStoreCheckpointInfo]]] = {
logInfo(log"Starting state rewrite for " +
log"checkpointLocation=${MDC(CHECKPOINT_LOCATION, resolvedCheckpointLocation)}, " +
log"readCheckpointLocation=" +
log"${MDC(CHECKPOINT_LOCATION, readResolvedCheckpointLocation.getOrElse(""))}, " +
log"readBatchId=${MDC(BATCH_ID, readBatchId)}, " +
log"writeBatchId=${MDC(BATCH_ID, writeBatchId)}")
- val (_, timeTakenMs) = Utils.timeTakenMs {
+ val (checkpointInfos, timeTakenMs) = Utils.timeTakenMs {
runInternal()
}
+ val allCheckpointInfos = checkpointInfos.values.flatMap { checkpointInfos =>
+ checkpointInfos.flatten.toSeq
Review Comment:
Since we are returning StateStoreCheckpointId, we don't need to do this conversion anymore
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-54590][SS] Support CheckpointV2 for StatePartitionAllColumnFamiliesWriter [spark]
Posted by "micheal-o (via GitHub)" <gi...@apache.org>.
micheal-o commented on PR #53720:
URL: https://github.com/apache/spark/pull/53720#issuecomment-3752852104
Also fix the PR title to: `Support Checkpoint V2 for State Rewriter and Repartitioning`
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-54590][SS] Support Checkpoint V2 for State Rewriter and Repartitioning [spark]
Posted by "anishshri-db (via GitHub)" <gi...@apache.org>.
anishshri-db commented on code in PR #53720:
URL: https://github.com/apache/spark/pull/53720#discussion_r2696023332
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/state/RocksDB.scala:
##########
@@ -443,77 +443,84 @@ class RocksDB(
private def loadWithCheckpointId(
version: Long,
stateStoreCkptId: Option[String],
- readOnly: Boolean = false): RocksDB = {
+ readOnly: Boolean = false,
+ loadEmpty: Boolean = false): RocksDB = {
// An array contains lineage information from [snapShotVersion, version]
// (inclusive in both ends)
var currVersionLineage: Array[LineageItem] = lineageManager.getLineageForCurrVersion()
try {
- if (loadedVersion != version || (loadedStateStoreCkptId.isEmpty ||
- stateStoreCkptId.get != loadedStateStoreCkptId.get)) {
+ if (loadEmpty || loadedVersion != version || loadedStateStoreCkptId.isEmpty ||
+ stateStoreCkptId.get != loadedStateStoreCkptId.get) {
closeDB(ignoreException = false)
-
- val (latestSnapshotVersion, latestSnapshotUniqueId) = {
- // Special handling when version is 0.
- // When loading the very first version (0), stateStoreCkptId does not need to be defined
- // because there won't be 0.changelog / 0.zip file created in RocksDB under v2.
- if (version == 0) {
- assert(stateStoreCkptId.isEmpty,
- "stateStoreCkptId should be empty when version is zero")
- (0L, None)
- // When there is a snapshot file, it is the ground truth, we can skip
- // reconstructing the lineage from changelog file.
- } else if (fileManager.existsSnapshotFile(version, stateStoreCkptId)) {
- currVersionLineage = Array(LineageItem(version, stateStoreCkptId.get))
- (version, stateStoreCkptId)
- } else {
- currVersionLineage = getLineageFromChangelogFile(version, stateStoreCkptId) :+
- LineageItem(version, stateStoreCkptId.get)
- currVersionLineage = currVersionLineage.sortBy(_.version)
-
- val latestSnapshotVersionsAndUniqueId =
- fileManager.getLatestSnapshotVersionAndUniqueIdFromLineage(currVersionLineage)
- latestSnapshotVersionsAndUniqueId match {
- case Some(pair) => (pair._1, Option(pair._2))
- case None if currVersionLineage.head.version == 1L =>
- logDebug(log"Cannot find latest snapshot based on lineage but first version " +
- log"is 1, use 0 as default. Lineage: ${MDC(LogKeys.LINEAGE, lineageManager)}")
- (0L, None)
- case _ =>
- throw QueryExecutionErrors.cannotFindBaseSnapshotCheckpoint(
- printLineageItems(currVersionLineage))
+ if (loadEmpty) {
+ require(stateStoreCkptId.isEmpty,
+ "stateStoreCkptId should be empty when loadEmpty is true")
+ loadEmptyStore(version)
+ lineageManager.clear()
+ } else {
+ val (latestSnapshotVersion, latestSnapshotUniqueId) = {
+ // Special handling when version is 0.
+ // When loading the very first version (0), stateStoreCkptId does not need to be defined
+ // because there won't be 0.changelog / 0.zip file created in RocksDB under v2.
+ if (version == 0) {
+ assert(stateStoreCkptId.isEmpty,
+ "stateStoreCkptId should be empty when version is zero")
+ (0L, None)
+ // When there is a snapshot file, it is the ground truth, we can skip
+ // reconstructing the lineage from changelog file.
+ } else if (fileManager.existsSnapshotFile(version, stateStoreCkptId)) {
+ currVersionLineage = Array(LineageItem(version, stateStoreCkptId.get))
+ (version, stateStoreCkptId)
+ } else {
+ currVersionLineage = getLineageFromChangelogFile(version, stateStoreCkptId) :+
+ LineageItem(version, stateStoreCkptId.get)
+ currVersionLineage = currVersionLineage.sortBy(_.version)
+
+ val latestSnapshotVersionsAndUniqueId =
+ fileManager.getLatestSnapshotVersionAndUniqueIdFromLineage(currVersionLineage)
+ latestSnapshotVersionsAndUniqueId match {
+ case Some(pair) => (pair._1, Option(pair._2))
+ case None if currVersionLineage.head.version == 1L =>
+ logDebug(log"Cannot find latest snapshot based on lineage but first version " +
+ log"is 1, use 0 as default. Lineage: ${MDC(LogKeys.LINEAGE, lineageManager)}")
+ (0L, None)
+ case _ =>
+ throw QueryExecutionErrors.cannotFindBaseSnapshotCheckpoint(
+ printLineageItems(currVersionLineage))
+ }
}
}
- }
- logInfo(log"Loaded latestSnapshotVersion: ${
- MDC(LogKeys.SNAPSHOT_VERSION, latestSnapshotVersion)}, latestSnapshotUniqueId: ${
- MDC(LogKeys.UUID, latestSnapshotUniqueId)}")
+ logInfo(log"Loaded latestSnapshotVersion: ${
Review Comment:
Is the indent as expected here ?
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/state/RocksDB.scala:
##########
@@ -530,9 +537,13 @@ class RocksDB(
if (conf.resetStatsOnLoad) {
nativeStats.reset
}
-
- logInfo(log"Loaded ${MDC(LogKeys.VERSION_NUM, version)} " +
- log"with uniqueId ${MDC(LogKeys.UUID, stateStoreCkptId)}")
+ if (loadEmpty) {
+ logInfo(log"Loaded empty store at version ${MDC(LogKeys.VERSION_NUM, version)} " +
+ log"with uniqueId")
Review Comment:
Is unique Id not available here ?
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/state/RocksDB.scala:
##########
@@ -443,77 +443,84 @@ class RocksDB(
private def loadWithCheckpointId(
version: Long,
stateStoreCkptId: Option[String],
- readOnly: Boolean = false): RocksDB = {
+ readOnly: Boolean = false,
+ loadEmpty: Boolean = false): RocksDB = {
// An array contains lineage information from [snapShotVersion, version]
// (inclusive in both ends)
var currVersionLineage: Array[LineageItem] = lineageManager.getLineageForCurrVersion()
try {
- if (loadedVersion != version || (loadedStateStoreCkptId.isEmpty ||
- stateStoreCkptId.get != loadedStateStoreCkptId.get)) {
+ if (loadEmpty || loadedVersion != version || loadedStateStoreCkptId.isEmpty ||
Review Comment:
Why do we need this ?
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/state/StateRewriter.scala:
##########
@@ -224,9 +267,8 @@ class StateRewriter(
schemaProvider,
executorSqlConf
)
-
- partitionWriter.write(partitionIter)
- }
+ Iterator(partitionWriter.write(partitionIter))
+ }.collect()
Review Comment:
Why are we calling `collect` here ?
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-54590][SS] Support CheckpointV2 for StatePartitionAllColumnFamiliesWriter [spark]
Posted by "micheal-o (via GitHub)" <gi...@apache.org>.
micheal-o commented on code in PR #53720:
URL: https://github.com/apache/spark/pull/53720#discussion_r2692923807
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/state/StateRewriter.scala:
##########
@@ -402,3 +477,15 @@ private[state] class StateRewriterUnsupportedStoreMetadataVersionError(
checkpointLocation,
subClass = "UNSUPPORTED_STATE_STORE_METADATA_VERSION",
messageParameters = Map.empty)
+
+private[state] class StateRewriterCheckpointVersionMismatchError(
Review Comment:
ditto: name
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/state/StateRewriter.scala:
##########
@@ -224,9 +261,8 @@ class StateRewriter(
schemaProvider,
executorSqlConf
)
-
- partitionWriter.write(partitionIter)
- }
+ Iterator(partitionWriter.write(partitionIter))
+ }.sortBy(_.partitionId).collect()
Review Comment:
qq, I thought the result is already sorted by partitionId? or that isn't true?
##########
common/utils/src/main/resources/error/error-conditions.json:
##########
@@ -5608,6 +5608,13 @@
"Unsupported state store metadata version encountered.",
"Only StateStoreMetadataV1 and StateStoreMetadataV2 are supported."
]
+ },
+ "CHECKPOINT_VERSION_MISMATCH" : {
Review Comment:
STATE_CHECKPOINT_FORMAT_VERSION_MISMATCH
##########
sql/core/src/test/scala/org/apache/spark/sql/execution/streaming/state/StatePartitionAllColumnFamiliesWriterSuite.scala:
##########
@@ -620,358 +957,42 @@ class StatePartitionAllColumnFamiliesWriterSuite extends StateDataSourceTestBase
}
}
}
+ } // End of foreach loop for changelog checkpointing dimension
- // Run transformWithState tests with different encoding formats
- Seq("unsaferow", "avro").foreach { encodingFormat =>
- def testWithChangelogAndEncodingConfig(testName: String)(testFun: => Unit): Unit = {
- test(s"$testName ($changelogCpTestSuffix, encoding = $encodingFormat)") {
- withSQLConf(
- "spark.sql.streaming.stateStore.rocksdb.changelogCheckpointing.enabled" ->
- changelogCheckpointingEnabled.toString,
- SQLConf.STREAMING_STATE_STORE_ENCODING_FORMAT.key -> encodingFormat) {
- testFun
- }
- }
- }
-
- testWithChangelogAndEncodingConfig(
- "SPARK-54411: transformWithState with multiple column families") {
- withTempDir { sourceDir =>
- withTempDir { targetDir =>
- val inputData = MemoryStream[String]
- val query = inputData.toDS()
- .groupByKey(x => x)
- .transformWithState(new MultiStateVarProcessor(),
- TimeMode.None(),
- OutputMode.Update())
- def runQuery(checkpointLocation: String, roundsOfData: Int): Unit = {
- val dataActions = (1 to roundsOfData).flatMap { _ =>
- Seq(
- AddData(inputData, "a", "b", "a"),
- ProcessAllAvailable()
- )
- }
- testStream(query)(
- Seq(StartStream(checkpointLocation = checkpointLocation)) ++
- dataActions ++
- Seq(StopStream): _*
- )
- }
-
- runQuery(sourceDir.getAbsolutePath, 2)
- runQuery(targetDir.getAbsolutePath, 1)
-
- val schemas = MultiStateVarProcessorTestUtils.getSchemasWithMetadata()
- val columnFamilyToSelectExprs = MultiStateVarProcessorTestUtils
- .getColumnFamilyToSelectExprs()
-
- val columnFamilyToStateSourceOptions = schemas.keys.map { cfName =>
- val base = Map(
- StateSourceOptions.STATE_VAR_NAME -> cfName
- )
-
- val withFlatten =
- if (cfName == MultiStateVarProcessorTestUtils.ITEMS_LIST) {
- base + (StateSourceOptions.FLATTEN_COLLECTION_TYPES -> "true")
- } else {
- base
- }
-
- cfName -> withFlatten
- }.toMap
-
- performRoundTripTest(
- sourceDir.getAbsolutePath,
- targetDir.getAbsolutePath,
- storeToColumnFamilies = Map(StateStoreId.DEFAULT_STORE_NAME -> schemas.keys.toList),
- storeToColumnFamilyToSelectExprs =
- Map(StateStoreId.DEFAULT_STORE_NAME -> columnFamilyToSelectExprs),
- storeToColumnFamilyToStateSourceOptions =
- Map(StateStoreId.DEFAULT_STORE_NAME -> columnFamilyToStateSourceOptions),
- operatorName = StatefulOperatorsUtils.TRANSFORM_WITH_STATE_EXEC_OP_NAME
- )
- }
- }
- }
-
- testWithChangelogAndEncodingConfig("SPARK-54411: transformWithState with eventTime timers") {
- withTempDir { sourceDir =>
- withTempDir { targetDir =>
- val inputData = MemoryStream[(String, Long)]
- val result = inputData.toDS()
- .select(col("_1").as("key"), timestamp_seconds(col("_2")).as("eventTime"))
- .withWatermark("eventTime", "10 seconds")
- .as[(String, Timestamp)]
- .groupByKey(_._1)
- .transformWithState(
- new EventTimeTimerProcessor(),
- TimeMode.EventTime(),
- OutputMode.Update())
-
- testStream(result, OutputMode.Update())(
- StartStream(checkpointLocation = sourceDir.getAbsolutePath),
- AddData(inputData, ("a", 1L), ("b", 2L), ("c", 3L)),
- ProcessAllAvailable(),
- StopStream
- )
-
- testStream(result, OutputMode.Update())(
- StartStream(checkpointLocation = targetDir.getAbsolutePath),
- AddData(inputData, ("x", 1L)),
- ProcessAllAvailable(),
- StopStream
- )
-
- val (schemaMap, selectExprs, stateSourceOptions) =
- getTimerStateConfigsForCountState(TimeMode.EventTime())
-
- performRoundTripTest(
- sourceDir.getAbsolutePath,
- targetDir.getAbsolutePath,
- storeToColumnFamilies =
- Map(StateStoreId.DEFAULT_STORE_NAME -> schemaMap.keys.toList),
- storeToColumnFamilyToSelectExprs =
- Map(StateStoreId.DEFAULT_STORE_NAME -> selectExprs),
- storeToColumnFamilyToStateSourceOptions =
- Map(StateStoreId.DEFAULT_STORE_NAME -> stateSourceOptions),
- operatorName = StatefulOperatorsUtils.TRANSFORM_WITH_STATE_EXEC_OP_NAME
- )
- }
- }
- }
-
- testWithChangelogAndEncodingConfig(
- "SPARK-54411: transformWithState with processing time timers") {
- withTempDir { sourceDir =>
- withTempDir { targetDir =>
- val clock = new StreamManualClock
- val inputData = MemoryStream[String]
- val result = inputData.toDS()
- .groupByKey(x => x)
- .transformWithState(new RunningCountStatefulProcessorWithProcTimeTimer(),
- TimeMode.ProcessingTime(),
- OutputMode.Update())
-
- testStream(result, OutputMode.Update())(
- StartStream(checkpointLocation = sourceDir.getAbsolutePath,
- trigger = Trigger.ProcessingTime("1 second"),
- triggerClock = clock),
- AddData(inputData, "a"),
- AdvanceManualClock(1 * 1000),
- CheckNewAnswer(("a", "1")),
- StopStream
- )
-
- val clock2 = new StreamManualClock
- testStream(result, OutputMode.Update())(
- StartStream(checkpointLocation = targetDir.getAbsolutePath,
- trigger = Trigger.ProcessingTime("1 second"),
- triggerClock = clock2),
- AddData(inputData, "x"),
- AdvanceManualClock(1 * 1000),
- CheckNewAnswer(("a", "1"), ("x", "1")),
- StopStream
- )
-
- val (schemaMap, selectExprs, sourceOptions) =
- getTimerStateConfigsForCountState(TimeMode.ProcessingTime())
-
- performRoundTripTest(
- sourceDir.getAbsolutePath,
- targetDir.getAbsolutePath,
- storeToColumnFamilies =
- Map(StateStoreId.DEFAULT_STORE_NAME -> schemaMap.keys.toList),
- storeToColumnFamilyToSelectExprs =
- Map(StateStoreId.DEFAULT_STORE_NAME -> selectExprs),
- storeToColumnFamilyToStateSourceOptions =
- Map(StateStoreId.DEFAULT_STORE_NAME -> sourceOptions),
- operatorName = StatefulOperatorsUtils.TRANSFORM_WITH_STATE_EXEC_OP_NAME
- )
- }
- }
- }
-
- testWithChangelogAndEncodingConfig("SPARK-54411: transformWithState with list and TTL") {
- withTempDir { sourceDir =>
- withTempDir { targetDir =>
- val clock = new StreamManualClock
- val inputData = MemoryStream[InputEvent]
- val ttlConfig = TTLConfig(ttlDuration = Duration.ofMinutes(1))
- val result = inputData.toDS()
- .groupByKey(x => x.key)
- .transformWithState(new ListStateTTLProcessor(ttlConfig),
- TimeMode.ProcessingTime(),
- OutputMode.Update())
-
- testStream(result, OutputMode.Update())(
- StartStream(checkpointLocation = sourceDir.getAbsolutePath,
- trigger = Trigger.ProcessingTime("1 second"),
- triggerClock = clock),
- AddData(inputData, InputEvent("k1", "put", 1)),
- AdvanceManualClock(1 * 1000),
- CheckNewAnswer(),
- StopStream
- )
-
- val clock2 = new StreamManualClock
- testStream(result, OutputMode.Update())(
- StartStream(checkpointLocation = targetDir.getAbsolutePath,
- trigger = Trigger.ProcessingTime("1 second"),
- triggerClock = clock2),
- AddData(inputData, InputEvent("k1", "append", 1)),
- AdvanceManualClock(1 * 1000),
- CheckNewAnswer(),
- StopStream
- )
-
- val schemas = TTLProcessorUtils.getListStateTTLSchemasWithMetadata()
-
- val columnFamilyToSelectExprs = Map(
- TTLProcessorUtils.LIST_STATE -> TTLProcessorUtils.getTTLSelectExpressions(
- TTLProcessorUtils.LIST_STATE
- ))
-
- val columnFamilyToStateSourceOptions = schemas.keys.map { cfName =>
- val base = Map(StateSourceOptions.STATE_VAR_NAME -> cfName)
-
- val withFlatten =
- if (cfName == TTLProcessorUtils.LIST_STATE) {
- base + (StateSourceOptions.FLATTEN_COLLECTION_TYPES -> "true")
- } else {
- base
- }
-
- cfName -> withFlatten
- }.toMap
-
- performRoundTripTest(
- sourceDir.getAbsolutePath,
- targetDir.getAbsolutePath,
- storeToColumnFamilies =
- Map(StateStoreId.DEFAULT_STORE_NAME -> schemas.keys.toList),
- storeToColumnFamilyToSelectExprs =
- Map(StateStoreId.DEFAULT_STORE_NAME -> columnFamilyToSelectExprs),
- storeToColumnFamilyToStateSourceOptions =
- Map(StateStoreId.DEFAULT_STORE_NAME -> columnFamilyToStateSourceOptions),
- operatorName = StatefulOperatorsUtils.TRANSFORM_WITH_STATE_EXEC_OP_NAME
- )
- }
- }
- }
-
- testWithChangelogAndEncodingConfig("SPARK-54411: transformWithState with map and TTL") {
- withTempDir { sourceDir =>
- withTempDir { targetDir =>
- val clock = new StreamManualClock
- val inputData = MemoryStream[MapInputEvent]
- val ttlConfig = TTLConfig(ttlDuration = Duration.ofMinutes(1))
- val result = inputData.toDS()
- .groupByKey(x => x.key)
- .transformWithState(new MapStateTTLProcessor(ttlConfig),
- TimeMode.ProcessingTime(),
- OutputMode.Update())
-
- testStream(result)(
- StartStream(checkpointLocation = sourceDir.getAbsolutePath,
- trigger = Trigger.ProcessingTime("1 second"),
- triggerClock = clock),
- AddData(inputData, MapInputEvent("a", "key1", "put", 1)),
- AdvanceManualClock(1 * 1000),
- CheckNewAnswer(),
- StopStream
- )
-
- val clock2 = new StreamManualClock
- testStream(result)(
- StartStream(checkpointLocation = targetDir.getAbsolutePath,
- trigger = Trigger.ProcessingTime("1 second"),
- triggerClock = clock2),
- AddData(inputData, MapInputEvent("x", "key1", "put", 1)),
- AdvanceManualClock(1 * 1000),
- CheckNewAnswer(),
- StopStream
- )
-
- val schemas = TTLProcessorUtils.getMapStateTTLSchemasWithMetadata()
-
- val columnFamilyToSelectExprs = Map(
- TTLProcessorUtils.MAP_STATE -> TTLProcessorUtils.getTTLSelectExpressions(
- TTLProcessorUtils.MAP_STATE
- ))
-
- val columnFamilyToStateSourceOptions = schemas.keys.map { cfName =>
- cfName -> Map(StateSourceOptions.STATE_VAR_NAME -> cfName)
- }.toMap
-
- performRoundTripTest(
- sourceDir.getAbsolutePath,
- targetDir.getAbsolutePath,
- storeToColumnFamilies =
- Map(StateStoreId.DEFAULT_STORE_NAME -> schemas.keys.toList),
- storeToColumnFamilyToSelectExprs =
- Map(StateStoreId.DEFAULT_STORE_NAME -> columnFamilyToSelectExprs),
- storeToColumnFamilyToStateSourceOptions =
- Map(StateStoreId.DEFAULT_STORE_NAME -> columnFamilyToStateSourceOptions),
- operatorName = StatefulOperatorsUtils.TRANSFORM_WITH_STATE_EXEC_OP_NAME
- )
+ test("SPARK-54590: Rewriter throw exception if checkpoint version is not set correct") {
+ withSQLConf(SQLConf.STATE_STORE_CHECKPOINT_FORMAT_VERSION.key -> "2") {
+ withTempDir { sourceDir =>
+ withTempDir { targetDir =>
+ // Step 1: Create state by running a streaming aggregation
+ runDropDuplicatesQuery(sourceDir.getAbsolutePath)
+ val sourceCheckpointMetadata = new StreamingQueryCheckpointMetadata(
+ spark, sourceDir.getAbsolutePath)
+ val readBatchId = sourceCheckpointMetadata.commitLog.getLatestBatchId().get
+ // Forced set STATE_STORE_CHECKPOINT_FORMAT_VERSION to 1 to mimic when user forgot to
+ // update checkpoint version to 2 in sqlConfig when running stateRewriter
+ // on checkpointV2 query.
+ spark.conf.unset(SQLConf.STATE_STORE_CHECKPOINT_FORMAT_VERSION.key)
+ val rewriter = new StateRewriter(
+ spark,
+ readBatchId,
+ readBatchId + 1,
+ sourceDir.getAbsolutePath,
+ spark.sessionState.newHadoopConf()
+ )
+ val ex = intercept[StateRewriterInvalidCheckpointError] {
+ rewriter.run()
}
- }
- }
-
- testWithChangelogAndEncodingConfig("SPARK-54411: transformWithState with value and TTL") {
- withTempDir { sourceDir =>
- withTempDir { targetDir =>
- val clock = new StreamManualClock
- val inputData = MemoryStream[InputEvent]
- val ttlConfig = TTLConfig(ttlDuration = Duration.ofMinutes(1))
- val result = inputData.toDS()
- .groupByKey(x => x.key)
- .transformWithState(new ValueStateTTLProcessor(ttlConfig),
- TimeMode.ProcessingTime(),
- OutputMode.Update())
-
- testStream(result)(
- StartStream(checkpointLocation = sourceDir.getAbsolutePath,
- trigger = Trigger.ProcessingTime("1 second"),
- triggerClock = clock),
- AddData(inputData, InputEvent("k1", "put", 1)),
- AddData(inputData, InputEvent("k2", "put", 2)),
- AdvanceManualClock(1 * 1000),
- CheckNewAnswer(),
- StopStream
- )
- val clock2 = new StreamManualClock
- testStream(result)(
- StartStream(checkpointLocation = targetDir.getAbsolutePath,
- trigger = Trigger.ProcessingTime("1 second"),
- triggerClock = clock2),
- AddData(inputData, InputEvent("x", "put", 1)),
- AdvanceManualClock(1 * 1000),
- CheckNewAnswer(),
- StopStream
- )
-
- val schemas = TTLProcessorUtils.getValueStateTTLSchemasWithMetadata()
-
- val columnFamilyToStateSourceOptions = schemas.keys.map { cfName =>
- cfName -> Map(StateSourceOptions.STATE_VAR_NAME -> cfName)
- }.toMap
-
- performRoundTripTest(
- sourceDir.getAbsolutePath,
- targetDir.getAbsolutePath,
- storeToColumnFamilies =
- Map(StateStoreId.DEFAULT_STORE_NAME -> schemas.keys.toList),
- storeToColumnFamilyToStateSourceOptions =
- Map(StateStoreId.DEFAULT_STORE_NAME -> columnFamilyToStateSourceOptions),
- operatorName = StatefulOperatorsUtils.TRANSFORM_WITH_STATE_EXEC_OP_NAME
- )
- }
+ assert(ex.getCondition == "STATE_REWRITER_INVALID_CHECKPOINT.CHECKPOINT_VERSION_MISMATCH")
Review Comment:
use `checkError` func instead of doing these asserts
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/state/StateRewriter.scala:
##########
@@ -342,6 +378,37 @@ class StateRewriter(
None
}
}
+
+ private def verifyCheckpointVersion(): Unit = {
+ // Verify checkpoint version in sqlConf based on commitLog for readCheckpoint
+ // in case user forgot to set STATE_STORE_CHECKPOINT_FORMAT_VERSION.
+ // Using read batch commit since the latest commit could be a skipped batch.
+ // If SQLConf.STATE_STORE_CHECKPOINT_FORMAT_VERSION is wrong, readCheckpoint.commitLog
+ // will throw an exception, and we will propagate this exception upstream.
+ // This prevents the StateRewriter from failing to write the correct state files
+ try {
+ val writeCheckpoint =
+ new StreamingQueryCheckpointMetadata(sparkSession, resolvedCheckpointLocation)
+ val readCheckpoint = if (readResolvedCheckpointLocation.isDefined) {
+ new StreamingQueryCheckpointMetadata(sparkSession, readResolvedCheckpointLocation.get)
+ } else {
+ // Same checkpoint for read & write
+ writeCheckpoint
+ }
+ readCheckpoint.commitLog.get(readBatchId)
+ } catch {
+ case e: IllegalStateException =>
Review Comment:
Do: `case e: IllegalStateException if e.getCause != null && e.getCause.isInstanceOf[SparkThrowable] =>`.
Otherwise the code below can lead to `NullPointerException`
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/state/StateRewriter.scala:
##########
@@ -81,24 +82,58 @@ class StateRewriter(
private val stateRootLocation = new Path(
resolvedCheckpointLocation, StreamingCheckpointConstants.DIR_NAME_STATE).toString
- def run(): Unit = {
+ // If checkpoint id is enabled, return
+ // Map[operatorId, Array[partition -> Array[stateStore -> StateStoreCheckpointId]]].
+ // Otherwise, return None
+ def run(): Option[Map[Long, Array[Array[String]]]] = {
logInfo(log"Starting state rewrite for " +
log"checkpointLocation=${MDC(CHECKPOINT_LOCATION, resolvedCheckpointLocation)}, " +
log"readCheckpointLocation=" +
log"${MDC(CHECKPOINT_LOCATION, readResolvedCheckpointLocation.getOrElse(""))}, " +
log"readBatchId=${MDC(BATCH_ID, readBatchId)}, " +
log"writeBatchId=${MDC(BATCH_ID, writeBatchId)}")
- val (_, timeTakenMs) = Utils.timeTakenMs {
+ val (checkpointIds, timeTakenMs) = Utils.timeTakenMs {
runInternal()
}
logInfo(log"State rewrite completed in ${MDC(DURATION, timeTakenMs)} ms for " +
- log"checkpointLocation=${MDC(CHECKPOINT_LOCATION, resolvedCheckpointLocation)}")
+ log"checkpointLocation=${MDC(CHECKPOINT_LOCATION, resolvedCheckpointLocation)} " +
+ log"checkpointInfos=${MDC(LAST_COMMITTED_CHECKPOINT_ID, checkpointIds)}"
Review Comment:
Actually we can skip logging the `checkpointIds` here, since they could be alot (e.g. 200 per operator) and would cause noise.
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/state/StateRewriter.scala:
##########
@@ -342,6 +378,37 @@ class StateRewriter(
None
}
}
+
+ private def verifyCheckpointVersion(): Unit = {
+ // Verify checkpoint version in sqlConf based on commitLog for readCheckpoint
+ // in case user forgot to set STATE_STORE_CHECKPOINT_FORMAT_VERSION.
+ // Using read batch commit since the latest commit could be a skipped batch.
+ // If SQLConf.STATE_STORE_CHECKPOINT_FORMAT_VERSION is wrong, readCheckpoint.commitLog
+ // will throw an exception, and we will propagate this exception upstream.
+ // This prevents the StateRewriter from failing to write the correct state files
+ try {
+ val writeCheckpoint =
Review Comment:
Please look at my previous comment [here](https://github.com/apache/spark/pull/53720#discussion_r2688023829) again and the sample code I gave there:
1. This isn't how I suggested `writeCheckpoint` to be implemented. It needs to use the passed in write checkpoint metadata i.e. `writeCheckpointMetadata.getOrElse(...)`
2. I also said `writeCheckpoint` & `readCheckpoint` to be class vals i.e. defined at class level as `private lazy val`
Please take a look at that comment again.
##########
common/utils/src/main/resources/error/error-conditions.json:
##########
@@ -5608,6 +5608,13 @@
"Unsupported state store metadata version encountered.",
"Only StateStoreMetadataV1 and StateStoreMetadataV2 are supported."
]
+ },
+ "CHECKPOINT_VERSION_MISMATCH" : {
+ "message" : [
+ "The checkpoint format version in SQLConf does not match the checkpoint version in the commit log.",
+ "Expected version v<expectedVersion>, but found v<actualVersion>.",
+ "Please set '<sqlConfKey>' to <expectedVersion> in your SQLConf before running StateRewriter."
Review Comment:
nit: `before retrying.` No need to say running StateRewriter since users don't know about it.
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/state/StateRewriter.scala:
##########
@@ -364,6 +431,14 @@ private[state] object StateRewriterErrors {
checkpointLocation: String): StateRewriterInvalidCheckpointError = {
new StateRewriterUnsupportedStoreMetadataVersionError(checkpointLocation)
}
+
+ def checkpointVersionMismatchError(
Review Comment:
ditto: name
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-54590][SS] Support Checkpoint V2 for State Rewriter and Repartitioning [spark]
Posted by "zifeif2 (via GitHub)" <gi...@apache.org>.
zifeif2 commented on code in PR #53720:
URL: https://github.com/apache/spark/pull/53720#discussion_r2696200119
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/state/RocksDB.scala:
##########
@@ -443,77 +443,84 @@ class RocksDB(
private def loadWithCheckpointId(
version: Long,
stateStoreCkptId: Option[String],
- readOnly: Boolean = false): RocksDB = {
+ readOnly: Boolean = false,
+ loadEmpty: Boolean = false): RocksDB = {
// An array contains lineage information from [snapShotVersion, version]
// (inclusive in both ends)
var currVersionLineage: Array[LineageItem] = lineageManager.getLineageForCurrVersion()
try {
- if (loadedVersion != version || (loadedStateStoreCkptId.isEmpty ||
- stateStoreCkptId.get != loadedStateStoreCkptId.get)) {
+ if (loadEmpty || loadedVersion != version || loadedStateStoreCkptId.isEmpty ||
Review Comment:
The reason why we need to support loadEmpty in loadWithCheckpointId in RocksDB is in repartition, we don't need to read previous data, that's why we need to add loadEmpty in RocksDB
I put loadEmpty in this if statement along with `loadedVersion != version || loadedStateStoreCkptId.isEmpty ||...` to reduce some duplicate code, but looks like it makes it harder to understand. I can refactor the code to make loadEmpty its separate block
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-54590][SS] Support Checkpoint V2 for State Rewriter and Repartitioning [spark]
Posted by "zifeif2 (via GitHub)" <gi...@apache.org>.
zifeif2 commented on code in PR #53720:
URL: https://github.com/apache/spark/pull/53720#discussion_r2696193184
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/state/StateRewriter.scala:
##########
@@ -224,9 +267,8 @@ class StateRewriter(
schemaProvider,
executorSqlConf
)
-
- partitionWriter.write(partitionIter)
- }
+ Iterator(partitionWriter.write(partitionIter))
+ }.collect()
Review Comment:
I thought we need to add collect() to make the rewrite actually happen and get a list of StateStoreCheckpointInfo
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-54590][SS] Support Checkpoint V2 for State Rewriter and Repartitioning [spark]
Posted by "zifeif2 (via GitHub)" <gi...@apache.org>.
zifeif2 commented on code in PR #53720:
URL: https://github.com/apache/spark/pull/53720#discussion_r2695507000
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/state/StateRewriter.scala:
##########
@@ -342,6 +378,37 @@ class StateRewriter(
None
}
}
+
+ private def verifyCheckpointVersion(): Unit = {
+ // Verify checkpoint version in sqlConf based on commitLog for readCheckpoint
+ // in case user forgot to set STATE_STORE_CHECKPOINT_FORMAT_VERSION.
+ // Using read batch commit since the latest commit could be a skipped batch.
+ // If SQLConf.STATE_STORE_CHECKPOINT_FORMAT_VERSION is wrong, readCheckpoint.commitLog
+ // will throw an exception, and we will propagate this exception upstream.
+ // This prevents the StateRewriter from failing to write the correct state files
+ try {
+ val writeCheckpoint =
Review Comment:
Sorry I missed the first point. Will address it in the next version.
For the second point, I am thinking if we should verify `CHECKPOINT_FORMAT_VERSION` as early as possible, in case future engineers use a lazily initialized `writeCheckpoint` and `readCheckpoint` in other functions without calling `verifyCheckpointFormatVersion`. In other words, we need to initialize `readCheckpoint` when constructing StateRewriter and call `verifyCheckpointFormatVersion` right after readCheckpoint is initialized, and we don't need to make `writeCheckpoint` and `readCheckpoint` lazy variables . What do you think?
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-54590][SS] Support Checkpoint V2 for State Rewriter and Repartitioning [spark]
Posted by "zifeif2 (via GitHub)" <gi...@apache.org>.
zifeif2 commented on code in PR #53720:
URL: https://github.com/apache/spark/pull/53720#discussion_r2696151840
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/state/RocksDB.scala:
##########
@@ -530,9 +537,13 @@ class RocksDB(
if (conf.resetStatsOnLoad) {
nativeStats.reset
}
-
- logInfo(log"Loaded ${MDC(LogKeys.VERSION_NUM, version)} " +
- log"with uniqueId ${MDC(LogKeys.UUID, stateStoreCkptId)}")
+ if (loadEmpty) {
+ logInfo(log"Loaded empty store at version ${MDC(LogKeys.VERSION_NUM, version)} " +
+ log"with uniqueId")
Review Comment:
Nope, we don't expect caller function to pass in uniqueId when they are calling loadWithCheckpointId when `loadEmpty = true`, because we are not load any previous versions of data when `loadEmpty=true`. We also have a require check above.
```
require(stateStoreCkptId.isEmpty, "stateStoreCkptId should be empty when loadEmpty is true")
```
I can change it to a less confusing message
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-54590][SS] Support CheckpointV2 for StatePartitionAllColumnFamiliesWriter [spark]
Posted by "micheal-o (via GitHub)" <gi...@apache.org>.
micheal-o commented on code in PR #53720:
URL: https://github.com/apache/spark/pull/53720#discussion_r2679160973
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/state/StateRewriter.scala:
##########
@@ -133,7 +136,7 @@ class StateRewriter(
val stateVarsIfTws = getStateVariablesIfTWS(opMetadata)
// Rewrite each state store of the operator
- stateStoresMetadata.foreach { stateStoreMetadata =>
+ opMetadata.operatorInfo.operatorId -> stateStoresMetadata.map { stateStoreMetadata =>
Review Comment:
nit: have a `val` here that collects the checkpointIds and then make this `operatorId -> checkpointIds` a separate line
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/state/RocksDB.scala:
##########
@@ -543,25 +548,30 @@ class RocksDB(
lineageManager.clear()
throw t
}
- if (enableChangelogCheckpointing && !readOnly) {
+ if (conf.enableChangelogCheckpointing && !readOnly) {
Review Comment:
Add this comment here too:
https://github.com/apache/spark/blob/acc80fd1b205792a43cc352a8e492e34bbe880da/sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/state/RocksDB.scala#L614
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/state/RocksDB.scala:
##########
@@ -543,25 +548,30 @@ class RocksDB(
lineageManager.clear()
throw t
}
- if (enableChangelogCheckpointing && !readOnly) {
+ if (conf.enableChangelogCheckpointing && !readOnly) {
// Make sure we don't leak resource.
changelogWriter.foreach(_.abort())
- // Initialize the changelog writer with lineage info
- // The lineage stored in changelog files should normally start with
- // the version of a snapshot, except for the first few versions.
- // Because they are solely loaded from changelog file.
- // (e.g. with default minDeltasForSnapshot, there is only 1_uuid1.changelog, no 1_uuid1.zip)
- // It should end with exactly one version before the change log's version.
- changelogWriter = Some(fileManager.getChangeLogWriter(
- version + 1,
- useColumnFamilies,
- sessionStateStoreCkptId,
- Some(currVersionLineage)))
+ if (loadEmpty) {
+ // No changelog writer for empty stores
Review Comment:
nit: "We don't want to write changelog file when loadEmpty is true"
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/state/StateRewriter.scala:
##########
@@ -81,23 +82,25 @@ class StateRewriter(
private val stateRootLocation = new Path(
resolvedCheckpointLocation, StreamingCheckpointConstants.DIR_NAME_STATE).toString
- def run(): Unit = {
+ // return a Map[operator id, Array[stateStore -> Array[partition -> StateStoreCheckpointInfo]]]
+ def run(): Map[Long, Array[Array[StateStoreCheckpointInfo]]] = {
logInfo(log"Starting state rewrite for " +
log"checkpointLocation=${MDC(CHECKPOINT_LOCATION, resolvedCheckpointLocation)}, " +
log"readCheckpointLocation=" +
log"${MDC(CHECKPOINT_LOCATION, readResolvedCheckpointLocation.getOrElse(""))}, " +
log"readBatchId=${MDC(BATCH_ID, readBatchId)}, " +
log"writeBatchId=${MDC(BATCH_ID, writeBatchId)}")
- val (_, timeTakenMs) = Utils.timeTakenMs {
+ val (checkpointInfos, timeTakenMs) = Utils.timeTakenMs {
runInternal()
}
logInfo(log"State rewrite completed in ${MDC(DURATION, timeTakenMs)} ms for " +
log"checkpointLocation=${MDC(CHECKPOINT_LOCATION, resolvedCheckpointLocation)}")
Review Comment:
lets also include the checkpointInfos in the log here
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/state/StateRewriter.scala:
##########
@@ -124,7 +127,7 @@ class StateRewriter(
// Do rewrite for each operator
// We can potentially parallelize this, but for now, do sequentially
- allOperatorsMetadata.foreach { opMetadata =>
+ allOperatorsMetadata.map { opMetadata =>
Review Comment:
Before we do rewrite, we need to check if the readBatchId has checkpointIds in the commitlog, if so, we should set `SQLConf.STATE_STORE_CHECKPOINT_FORMAT_VERSION` in the `sqlConf` to enable checkpoint v2.
We can't rely on the user to correctly set this conf in the session before running repartition. They may forget and this conf isn't part of the confs written to checkpoint. You can repro this, by not setting the conf in the session and the rewrite will fail. Lets also have a test to validate this.
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/state/OfflineStateRepartitionRunner.scala:
##########
@@ -289,12 +292,32 @@ class OfflineStateRepartitionRunner(
}
}
- private def commitBatch(newBatchId: Long, lastCommittedBatchId: Long): Unit = {
+ private def commitBatch(
+ newBatchId: Long,
+ lastCommittedBatchId: Long,
+ opIdToStateStoreCkptInfo: Option[Map[Long, Array[Array[StateStoreCheckpointInfo]]]]): Unit = {
val latestCommit = checkpointMetadata.commitLog.get(lastCommittedBatchId).get
-
- // todo: For checkpoint v2, we need to update the stateUniqueIds based on the
- // newly created state commit. Will be done in subsequent PR.
- if (!checkpointMetadata.commitLog.add(newBatchId, latestCommit)) {
+ val commitMetadata = opIdToStateStoreCkptInfo.map {originalInfoMap =>
+ // opIdToStateStoreCkptInfo is Map[operatorId, Array[stateStore -> Array[partition -> info]]]
+ // we change it to Map[operatorId, Array[partitionId -> Array[store -> info]]]
+ val opIdToPartitionCkptInfo: Map[Long, Array[Array[String]]] =
+ originalInfoMap.map {
+ case(operator, storesSeq) =>
+ val numPartitions = storesSeq.head.length
+ operator -> (0 until numPartitions).map { partitionIdx =>
+ storesSeq.flatMap { storePartitions =>
+ storePartitions(partitionIdx).stateStoreCkptId
+ }
+ }.toArray
+ }
+ // Include checkpoint IDs in commit metadata
+ CommitMetadata(
Review Comment:
nit: can just do `latestCommit.copy(stateUniqueIds = Option(opIdToPartitionCkptInfo))`
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/state/RocksDB.scala:
##########
@@ -443,77 +443,82 @@ class RocksDB(
private def loadWithCheckpointId(
version: Long,
stateStoreCkptId: Option[String],
- readOnly: Boolean = false): RocksDB = {
+ readOnly: Boolean = false,
+ loadEmpty: Boolean = false): RocksDB = {
// An array contains lineage information from [snapShotVersion, version]
// (inclusive in both ends)
var currVersionLineage: Array[LineageItem] = lineageManager.getLineageForCurrVersion()
try {
- if (loadedVersion != version || (loadedStateStoreCkptId.isEmpty ||
- stateStoreCkptId.get != loadedStateStoreCkptId.get)) {
+ if (loadEmpty || loadedVersion != version || loadedStateStoreCkptId.isEmpty ||
+ stateStoreCkptId.get != loadedStateStoreCkptId.get) {
closeDB(ignoreException = false)
-
- val (latestSnapshotVersion, latestSnapshotUniqueId) = {
- // Special handling when version is 0.
- // When loading the very first version (0), stateStoreCkptId does not need to be defined
- // because there won't be 0.changelog / 0.zip file created in RocksDB under v2.
- if (version == 0) {
- assert(stateStoreCkptId.isEmpty,
- "stateStoreCkptId should be empty when version is zero")
- (0L, None)
- // When there is a snapshot file, it is the ground truth, we can skip
- // reconstructing the lineage from changelog file.
- } else if (fileManager.existsSnapshotFile(version, stateStoreCkptId)) {
- currVersionLineage = Array(LineageItem(version, stateStoreCkptId.get))
- (version, stateStoreCkptId)
- } else {
- currVersionLineage = getLineageFromChangelogFile(version, stateStoreCkptId) :+
- LineageItem(version, stateStoreCkptId.get)
- currVersionLineage = currVersionLineage.sortBy(_.version)
-
- val latestSnapshotVersionsAndUniqueId =
- fileManager.getLatestSnapshotVersionAndUniqueIdFromLineage(currVersionLineage)
- latestSnapshotVersionsAndUniqueId match {
- case Some(pair) => (pair._1, Option(pair._2))
- case None if currVersionLineage.head.version == 1L =>
- logDebug(log"Cannot find latest snapshot based on lineage but first version " +
- log"is 1, use 0 as default. Lineage: ${MDC(LogKeys.LINEAGE, lineageManager)}")
- (0L, None)
- case _ =>
- throw QueryExecutionErrors.cannotFindBaseSnapshotCheckpoint(
- printLineageItems(currVersionLineage))
+ if (loadEmpty) {
+ loadEmptyStore(version)
+ lineageManager.clear()
+ } else {
+ val (latestSnapshotVersion, latestSnapshotUniqueId) = {
+ // Special handling when version is 0.
+ // When loading the very first version (0), stateStoreCkptId does not need to be defined
+ // because there won't be 0.changelog / 0.zip file created in RocksDB under v2.
+ if (version == 0) {
+ assert(stateStoreCkptId.isEmpty,
+ "stateStoreCkptId should be empty when version is zero")
+ (0L, None)
+ // When there is a snapshot file, it is the ground truth, we can skip
+ // reconstructing the lineage from changelog file.
+ } else if (fileManager.existsSnapshotFile(version, stateStoreCkptId)) {
+ currVersionLineage = Array(LineageItem(version, stateStoreCkptId.get))
+ (version, stateStoreCkptId)
+ } else {
+ currVersionLineage = getLineageFromChangelogFile(version, stateStoreCkptId) :+
+ LineageItem(version, stateStoreCkptId.get)
+ currVersionLineage = currVersionLineage.sortBy(_.version)
+
+ val latestSnapshotVersionsAndUniqueId =
+ fileManager.getLatestSnapshotVersionAndUniqueIdFromLineage(currVersionLineage)
+ latestSnapshotVersionsAndUniqueId match {
+ case Some(pair) => (pair._1, Option(pair._2))
+ case None if currVersionLineage.head.version == 1L =>
+ logDebug(log"Cannot find latest snapshot based on lineage but first version " +
+ log"is 1, use 0 as default. Lineage: ${MDC(LogKeys.LINEAGE, lineageManager)}")
+ (0L, None)
+ case _ =>
+ throw QueryExecutionErrors.cannotFindBaseSnapshotCheckpoint(
+ printLineageItems(currVersionLineage))
+ }
}
}
- }
- logInfo(log"Loaded latestSnapshotVersion: ${
- MDC(LogKeys.SNAPSHOT_VERSION, latestSnapshotVersion)}, latestSnapshotUniqueId: ${
- MDC(LogKeys.UUID, latestSnapshotUniqueId)}")
+ logInfo(log"Loaded latestSnapshotVersion: ${
+ MDC(LogKeys.SNAPSHOT_VERSION, latestSnapshotVersion)}, latestSnapshotUniqueId: ${
+ MDC(LogKeys.UUID, latestSnapshotUniqueId)}")
- val metadata = fileManager.loadCheckpointFromDfs(latestSnapshotVersion,
- workingDir, rocksDBFileMapping, latestSnapshotUniqueId)
+ val metadata = fileManager.loadCheckpointFromDfs(latestSnapshotVersion,
+ workingDir, rocksDBFileMapping, latestSnapshotUniqueId)
- loadedVersion = latestSnapshotVersion
+ loadedVersion = latestSnapshotVersion
- // reset the last snapshot version to the latest available snapshot version
- lastSnapshotVersion = latestSnapshotVersion
- lineageManager.resetLineage(currVersionLineage)
+ // reset the last snapshot version to the latest available snapshot version
+ lastSnapshotVersion = latestSnapshotVersion
+ lineageManager.resetLineage(currVersionLineage)
- // Initialize maxVersion upon successful load from DFS
- fileManager.setMaxSeenVersion(version)
+ // Initialize maxVersion upon successful load from DFS
+ fileManager.setMaxSeenVersion(version)
- // Report this snapshot version to the coordinator
- reportSnapshotUploadToCoordinator(latestSnapshotVersion)
+ // Report this snapshot version to the coordinator
+ reportSnapshotUploadToCoordinator(latestSnapshotVersion)
- openLocalRocksDB(metadata)
+ openLocalRocksDB(metadata)
- if (loadedVersion != version) {
- val versionsAndUniqueIds = currVersionLineage.collect {
+ if (loadedVersion != version) {
+ val versionsAndUniqueIds = currVersionLineage.collect {
case i if i.version > loadedVersion && i.version <= version =>
(i.version, Option(i.checkpointUniqueId))
}
- replayChangelog(versionsAndUniqueIds)
- loadedVersion = version
- lineageManager.resetLineage(currVersionLineage)
+ replayChangelog(versionsAndUniqueIds)
+ loadedVersion = version
+ lineageManager.resetLineage(currVersionLineage)
+ }
Review Comment:
should we update the `loaded version..` message below, just like we did for `loadWithoutCheckpointId`
https://github.com/apache/spark/blob/acc80fd1b205792a43cc352a8e492e34bbe880da/sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/state/RocksDB.scala#L604
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/state/OfflineStateRepartitionRunner.scala:
##########
@@ -289,12 +292,32 @@ class OfflineStateRepartitionRunner(
}
}
- private def commitBatch(newBatchId: Long, lastCommittedBatchId: Long): Unit = {
+ private def commitBatch(
+ newBatchId: Long,
+ lastCommittedBatchId: Long,
+ opIdToStateStoreCkptInfo: Option[Map[Long, Array[Array[StateStoreCheckpointInfo]]]]): Unit = {
val latestCommit = checkpointMetadata.commitLog.get(lastCommittedBatchId).get
-
- // todo: For checkpoint v2, we need to update the stateUniqueIds based on the
- // newly created state commit. Will be done in subsequent PR.
- if (!checkpointMetadata.commitLog.add(newBatchId, latestCommit)) {
+ val commitMetadata = opIdToStateStoreCkptInfo.map {originalInfoMap =>
+ // opIdToStateStoreCkptInfo is Map[operatorId, Array[stateStore -> Array[partition -> info]]]
+ // we change it to Map[operatorId, Array[partitionId -> Array[store -> info]]]
Review Comment:
Why not let state rewriter do this conversion before returning it. Otherwise every caller of state rewriter would need to implement this conversion
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/state/OfflineStateRepartitionRunner.scala:
##########
@@ -95,12 +95,14 @@ class OfflineStateRepartitionRunner(
transformFunc = Some(stateRepartitionFunc),
writeCheckpointMetadata = Some(checkpointMetadata)
)
- rewriter.run()
+ val operatorToCkptIds = rewriter.run()
updateNumPartitionsInOperatorMetadata(newBatchId, readBatchId = lastCommittedBatchId)
// Commit the repartition batch
- commitBatch(newBatchId, lastCommittedBatchId)
+ val enableCheckpointId = sparkSession.conf.get(
+ SQLConf.STATE_STORE_CHECKPOINT_FORMAT_VERSION.key).toInt >= 2
Review Comment:
Use `StatefulOperatorStateInfo.enableStateStoreCheckpointIds` util function for this. Also we can move this check into `commitBatch` func
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/state/RocksDB.scala:
##########
@@ -443,77 +443,82 @@ class RocksDB(
private def loadWithCheckpointId(
version: Long,
stateStoreCkptId: Option[String],
- readOnly: Boolean = false): RocksDB = {
+ readOnly: Boolean = false,
+ loadEmpty: Boolean = false): RocksDB = {
// An array contains lineage information from [snapShotVersion, version]
// (inclusive in both ends)
var currVersionLineage: Array[LineageItem] = lineageManager.getLineageForCurrVersion()
try {
- if (loadedVersion != version || (loadedStateStoreCkptId.isEmpty ||
- stateStoreCkptId.get != loadedStateStoreCkptId.get)) {
+ if (loadEmpty || loadedVersion != version || loadedStateStoreCkptId.isEmpty ||
+ stateStoreCkptId.get != loadedStateStoreCkptId.get) {
closeDB(ignoreException = false)
-
- val (latestSnapshotVersion, latestSnapshotUniqueId) = {
- // Special handling when version is 0.
- // When loading the very first version (0), stateStoreCkptId does not need to be defined
- // because there won't be 0.changelog / 0.zip file created in RocksDB under v2.
- if (version == 0) {
- assert(stateStoreCkptId.isEmpty,
- "stateStoreCkptId should be empty when version is zero")
- (0L, None)
- // When there is a snapshot file, it is the ground truth, we can skip
- // reconstructing the lineage from changelog file.
- } else if (fileManager.existsSnapshotFile(version, stateStoreCkptId)) {
- currVersionLineage = Array(LineageItem(version, stateStoreCkptId.get))
- (version, stateStoreCkptId)
- } else {
- currVersionLineage = getLineageFromChangelogFile(version, stateStoreCkptId) :+
- LineageItem(version, stateStoreCkptId.get)
- currVersionLineage = currVersionLineage.sortBy(_.version)
-
- val latestSnapshotVersionsAndUniqueId =
- fileManager.getLatestSnapshotVersionAndUniqueIdFromLineage(currVersionLineage)
- latestSnapshotVersionsAndUniqueId match {
- case Some(pair) => (pair._1, Option(pair._2))
- case None if currVersionLineage.head.version == 1L =>
- logDebug(log"Cannot find latest snapshot based on lineage but first version " +
- log"is 1, use 0 as default. Lineage: ${MDC(LogKeys.LINEAGE, lineageManager)}")
- (0L, None)
- case _ =>
- throw QueryExecutionErrors.cannotFindBaseSnapshotCheckpoint(
- printLineageItems(currVersionLineage))
+ if (loadEmpty) {
Review Comment:
require that `stateStoreCkptId` is not present
##########
sql/core/src/test/scala/org/apache/spark/sql/execution/streaming/state/OfflineStateRepartitionSuite.scala:
##########
@@ -106,71 +106,87 @@ class OfflineStateRepartitionSuite extends StreamTest
)
}
- test("Repartition: success, failure, retry") {
- withTempDir { dir =>
- val originalPartitions = 3
- val input = MemoryStream[Int]
- val batchId = runSimpleStreamQuery(originalPartitions, dir.getAbsolutePath, input)
- val checkpointMetadata = new StreamingQueryCheckpointMetadata(spark, dir.getAbsolutePath)
- // Shouldn't be seen as a repartition batch
- assert(!isRepartitionBatch(batchId, checkpointMetadata.offsetLog, dir.getAbsolutePath))
-
- // Trying to repartition to the same number should fail
- val ex = intercept[StateRepartitionShufflePartitionsAlreadyMatchError] {
- spark.streamingCheckpointManager.repartition(dir.getAbsolutePath, originalPartitions)
+ Seq(1, 2).foreach { ckptVersion =>
+ def testWithChangelogAndCheckpointId(testName: String)(testFun: => Unit): Unit = {
Review Comment:
nit: you mean `testWithCheckpointId`?
##########
sql/core/src/test/scala/org/apache/spark/sql/execution/streaming/state/OfflineStateRepartitionSuite.scala:
##########
@@ -106,71 +106,87 @@ class OfflineStateRepartitionSuite extends StreamTest
)
}
- test("Repartition: success, failure, retry") {
- withTempDir { dir =>
- val originalPartitions = 3
- val input = MemoryStream[Int]
- val batchId = runSimpleStreamQuery(originalPartitions, dir.getAbsolutePath, input)
- val checkpointMetadata = new StreamingQueryCheckpointMetadata(spark, dir.getAbsolutePath)
- // Shouldn't be seen as a repartition batch
- assert(!isRepartitionBatch(batchId, checkpointMetadata.offsetLog, dir.getAbsolutePath))
-
- // Trying to repartition to the same number should fail
- val ex = intercept[StateRepartitionShufflePartitionsAlreadyMatchError] {
- spark.streamingCheckpointManager.repartition(dir.getAbsolutePath, originalPartitions)
+ Seq(1, 2).foreach { ckptVersion =>
+ def testWithChangelogAndCheckpointId(testName: String)(testFun: => Unit): Unit = {
+ test(s"$testName (enableCkptId = ${ckptVersion >= 2})") {
+ withSQLConf(
+ SQLConf.STATE_STORE_CHECKPOINT_FORMAT_VERSION.key -> ckptVersion.toString) {
+ testFun
+ }
}
- checkError(
- ex,
- condition = "STATE_REPARTITION_INVALID_CHECKPOINT.SHUFFLE_PARTITIONS_ALREADY_MATCH",
- parameters = Map(
- "checkpointLocation" -> dir.getAbsolutePath,
- "batchId" -> batchId.toString,
- "numPartitions" -> originalPartitions.toString
- )
- )
-
- // Trying to repartition to a different number should succeed
- val newPartitions = originalPartitions + 1
- spark.streamingCheckpointManager.repartition(dir.getAbsolutePath, newPartitions)
- val repartitionBatchId = batchId + 1
- val hadoopConf = spark.sessionState.newHadoopConf()
- verifyRepartitionBatch(
- repartitionBatchId, checkpointMetadata, hadoopConf, dir.getAbsolutePath, newPartitions)
-
- // Now delete the repartition commit to simulate a failed repartition attempt.
- // This will delete all the commits after the batchId.
- checkpointMetadata.commitLog.purgeAfter(batchId)
+ }
- // Try to repartition with a different numPartitions should fail,
- // since it will see an uncommitted repartition batch with a different numPartitions.
- val ex2 = intercept[StateRepartitionLastBatchAbandonedRepartitionError] {
- spark.streamingCheckpointManager.repartition(dir.getAbsolutePath, newPartitions + 1)
- }
- checkError(
- ex2,
- condition = "STATE_REPARTITION_INVALID_CHECKPOINT.LAST_BATCH_ABANDONED_REPARTITION",
- parameters = Map(
- "checkpointLocation" -> dir.getAbsolutePath,
- "lastBatchId" -> repartitionBatchId.toString,
- "lastBatchShufflePartitions" -> newPartitions.toString,
- "numPartitions" -> (newPartitions + 1).toString
+ testWithChangelogAndCheckpointId("Repartition: success, failure, retry") {
+ withTempDir { dir =>
+ val originalPartitions = 3
+ val input = MemoryStream[Int]
+ val batchId = runSimpleStreamQuery(originalPartitions, dir.getAbsolutePath, input)
+ val checkpointMetadata = new StreamingQueryCheckpointMetadata(spark, dir.getAbsolutePath)
+ // Shouldn't be seen as a repartition batch
+ assert(!isRepartitionBatch(batchId, checkpointMetadata.offsetLog, dir.getAbsolutePath))
+
+ // Trying to repartition to the same number should fail
+ val ex = intercept[StateRepartitionShufflePartitionsAlreadyMatchError] {
+ spark.streamingCheckpointManager.repartition(dir.getAbsolutePath, originalPartitions)
+ }
+ checkError(
+ ex,
+ condition = "STATE_REPARTITION_INVALID_CHECKPOINT.SHUFFLE_PARTITIONS_ALREADY_MATCH",
+ parameters = Map(
+ "checkpointLocation" -> dir.getAbsolutePath,
+ "batchId" -> batchId.toString,
+ "numPartitions" -> originalPartitions.toString
+ )
)
- )
- // Retrying with the same numPartitions should work
- spark.streamingCheckpointManager.repartition(dir.getAbsolutePath, newPartitions)
- verifyRepartitionBatch(
- repartitionBatchId, checkpointMetadata, hadoopConf, dir.getAbsolutePath, newPartitions)
+ // Trying to repartition to a different number should succeed
+ val newPartitions = originalPartitions + 1
+ spark.streamingCheckpointManager.repartition(dir.getAbsolutePath, newPartitions)
+ val repartitionBatchId = batchId + 1
+ val hadoopConf = spark.sessionState.newHadoopConf()
+ verifyRepartitionBatch(
+ repartitionBatchId, checkpointMetadata, hadoopConf, dir.getAbsolutePath, newPartitions)
+
+ // Now delete the repartition commit to simulate a failed repartition attempt.
+ // This will delete all the commits after the batchId.
+ checkpointMetadata.commitLog.purgeAfter(batchId)
+
+ // Try to repartition with a different numPartitions should fail,
+ // since it will see an uncommitted repartition batch with a different numPartitions.
+ val ex2 = intercept[StateRepartitionLastBatchAbandonedRepartitionError] {
+ spark.streamingCheckpointManager.repartition(dir.getAbsolutePath, newPartitions + 1)
+ }
+ checkError(
+ ex2,
+ condition = "STATE_REPARTITION_INVALID_CHECKPOINT.LAST_BATCH_ABANDONED_REPARTITION",
+ parameters = Map(
+ "checkpointLocation" -> dir.getAbsolutePath,
+ "lastBatchId" -> repartitionBatchId.toString,
+ "lastBatchShufflePartitions" -> newPartitions.toString,
+ "numPartitions" -> (newPartitions + 1).toString
+ )
+ )
- // Repartition with way more partitions, to verify that empty partitions are properly created
- val morePartitions = newPartitions * 3
- spark.streamingCheckpointManager.repartition(dir.getAbsolutePath, morePartitions)
- verifyRepartitionBatch(
- repartitionBatchId + 1, checkpointMetadata, hadoopConf,
- dir.getAbsolutePath, morePartitions)
+ // Retrying with the same numPartitions should work
+ spark.streamingCheckpointManager.repartition(dir.getAbsolutePath, newPartitions)
+ verifyRepartitionBatch(
+ repartitionBatchId, checkpointMetadata, hadoopConf, dir.getAbsolutePath, newPartitions)
+
+ // Repartition with way more partitions, to verify that empty partitions are properly
+ // created
+ val morePartitions = newPartitions * 3
+ spark.streamingCheckpointManager.repartition(dir.getAbsolutePath, morePartitions)
+ verifyRepartitionBatch(
+ repartitionBatchId + 1, checkpointMetadata, hadoopConf,
+ dir.getAbsolutePath, morePartitions)
+ if (spark.conf.get(
+ SQLConf.STATE_STORE_CHECKPOINT_FORMAT_VERSION.key).toInt >= 2) {
+ verifyCheckpointIds(repartitionBatchId + 1, checkpointMetadata, morePartitions)
+ }
- // Restart the query to make sure it can start after repartitioning
- runSimpleStreamQuery(morePartitions, dir.getAbsolutePath, input)
+ // Restart the query to make sure it can start after repartitioning
+ runSimpleStreamQuery(morePartitions, dir.getAbsolutePath, input)
+ }
}
}
Review Comment:
we also need to run the two other test cases below with checkpoint id enabled/disabled
##########
sql/core/src/test/scala/org/apache/spark/sql/execution/streaming/state/StatePartitionAllColumnFamiliesWriterSuite.scala:
##########
@@ -450,8 +464,28 @@ class StatePartitionAllColumnFamiliesWriterSuite extends StateDataSourceTestBase
}
}
- testWithChangelogConfig("SPARK-54420: aggregation state ver 1") {
- testRoundTripForAggrStateVersion(1)
+ // Run transformWithState tests with enable/disable checkpoint V2
+ Seq(1, 2).foreach { ckptVersion =>
+ def testWithChangelogAndCheckpointId(testName: String)(testFun: => Unit): Unit = {
+ test(s"$testName ($changelogCpTestSuffix, enableCkptId = ${ckptVersion >= 2})") {
+ withSQLConf(
+ "spark.sql.streaming.stateStore.rocksdb.changelogCheckpointing.enabled" ->
+ changelogCheckpointingEnabled.toString,
+ SQLConf.STATE_STORE_CHECKPOINT_FORMAT_VERSION.key -> ckptVersion.toString) {
+ testFun
+ }
+ }
+ }
+
+ testWithChangelogAndCheckpointId("SPARK-54420: aggregation state ver 1") {
+ testRoundTripForAggrStateVersion(1)
+ }
+
+ Seq(1, 2).foreach { version =>
+ testWithChangelogAndCheckpointId(s"SPARK-54420: stream-stream join state ver $version") {
+ testStreamStreamJoinRoundTrip(version)
+ }
+ }
Review Comment:
Shouldn't we be testing checkpoint v2 for the others too. At least we should also test for TWS too.
##########
sql/core/src/test/scala/org/apache/spark/sql/execution/streaming/state/StatePartitionAllColumnFamiliesWriterSuite.scala:
##########
@@ -95,11 +96,35 @@ class StatePartitionAllColumnFamiliesWriterSuite extends StateDataSourceTestBase
transformFunc = None,
writeCheckpointMetadata = Some(targetCheckpointMetadata)
)
- rewriter.run()
+ val checkpointInfos = rewriter.run()
- // Commit to commitLog
+ // Commit to commitLog with checkpoint IDs
val latestCommit = targetCheckpointMetadata.commitLog.get(lastBatch).get
- targetCheckpointMetadata.commitLog.add(writeBatchId, latestCommit)
+ val operatorId = 0L
+ val storesSeq: Array[Array[StateStoreCheckpointInfo]] = checkpointInfos(operatorId)
+ val commitMetadata = if (StateStoreConf(conf).enableStateStoreCheckpointIds) {
Review Comment:
We won't need this conversion here if rewriter is doing the conversion before returning it. It is cleaner since each rewriter caller won't need to implement this.
##########
sql/core/src/test/scala/org/apache/spark/sql/execution/streaming/state/StatePartitionAllColumnFamiliesWriterSuite.scala:
##########
@@ -95,11 +96,35 @@ class StatePartitionAllColumnFamiliesWriterSuite extends StateDataSourceTestBase
transformFunc = None,
writeCheckpointMetadata = Some(targetCheckpointMetadata)
)
- rewriter.run()
+ val checkpointInfos = rewriter.run()
- // Commit to commitLog
+ // Commit to commitLog with checkpoint IDs
val latestCommit = targetCheckpointMetadata.commitLog.get(lastBatch).get
- targetCheckpointMetadata.commitLog.add(writeBatchId, latestCommit)
+ val operatorId = 0L
+ val storesSeq: Array[Array[StateStoreCheckpointInfo]] = checkpointInfos(operatorId)
+ val commitMetadata = if (StateStoreConf(conf).enableStateStoreCheckpointIds) {
+ // Build map: partition id -> array of checkpoint IDs (one per store, in order)
+ // checkpointInfos(operatorId) is Seq[Array[StateStoreCheckpointInfo]]
+ // where Seq is stores (in order: store0, store1, ...), Array is partitions
+ // For join operators: 4 stores (left-keyToNumValues, left-keyWithIndexToValue,
+ // right-keyToNumValues, right-keyWithIndexToValue)
+ // For regular operators: 1 store
+ val numPartitions = storesSeq.head.length
+ val ckptIds: Array[Array[String]] = (0 until numPartitions).map { partitionIdx =>
+ // For this partition, collect checkpoint IDs from all stores (in order)
+ storesSeq.flatMap { storePartitions =>
+ storePartitions(partitionIdx).stateStoreCkptId
+ }
+ }.toArray
+ // Include checkpoint IDs in commit metadata
+ CommitMetadata(
Review Comment:
ditto, use copy() instead
##########
sql/core/src/test/scala/org/apache/spark/sql/execution/streaming/state/OfflineStateRepartitionSuite.scala:
##########
@@ -454,4 +470,40 @@ object OfflineStateRepartitionTestUtils {
}
}
}
+
+ def verifyCheckpointIds(
+ repartitionBatchId: Long,
+ checkpointMetadata: StreamingQueryCheckpointMetadata,
+ expectedShufflePartitions: Int): Unit = {
+ // Verify commit log has the repartition batch with checkpoint IDs
+ val commitOpt = checkpointMetadata.commitLog.get(repartitionBatchId)
+ assert(commitOpt.isDefined, s"Commit for batch $repartitionBatchId should exist")
+
+ val commitMetadata = commitOpt.get
+
+ // Verify stateUniqueIds is present for checkpoint V2
+ assert(commitMetadata.stateUniqueIds.isDefined,
+ "stateUniqueIds should be present in commit metadata when checkpoint version >= 2")
+
+ val operatorIdToCkptInfos = commitMetadata.stateUniqueIds.get
+ assert(operatorIdToCkptInfos.nonEmpty,
+ "operatorIdToCkptInfos should not be empty")
+
+ // Verify structure for each operator
+ operatorIdToCkptInfos.foreach { case (operatorId, partitionToCkptIds) =>
+ // Should have checkpoint IDs for all partitions
+ assert(partitionToCkptIds.length == expectedShufflePartitions,
+ s"Operator $operatorId: Expected $expectedShufflePartitions partition checkpoint IDs, " +
+ s"but found ${partitionToCkptIds.length}")
+ // Each partition should have checkpoint IDs (at least one per store)
+ partitionToCkptIds.foreach { ckptIds =>
+ assert(ckptIds.nonEmpty, s"Operator $operatorId should have checkpoint IDs")
Review Comment:
nit: include the partitionId in the message right?
##########
sql/core/src/test/scala/org/apache/spark/sql/execution/streaming/state/OfflineStateRepartitionSuite.scala:
##########
@@ -454,4 +470,40 @@ object OfflineStateRepartitionTestUtils {
}
}
}
+
+ def verifyCheckpointIds(
+ repartitionBatchId: Long,
+ checkpointMetadata: StreamingQueryCheckpointMetadata,
+ expectedShufflePartitions: Int): Unit = {
+ // Verify commit log has the repartition batch with checkpoint IDs
+ val commitOpt = checkpointMetadata.commitLog.get(repartitionBatchId)
+ assert(commitOpt.isDefined, s"Commit for batch $repartitionBatchId should exist")
+
+ val commitMetadata = commitOpt.get
+
+ // Verify stateUniqueIds is present for checkpoint V2
+ assert(commitMetadata.stateUniqueIds.isDefined,
+ "stateUniqueIds should be present in commit metadata when checkpoint version >= 2")
+
+ val operatorIdToCkptInfos = commitMetadata.stateUniqueIds.get
+ assert(operatorIdToCkptInfos.nonEmpty,
+ "operatorIdToCkptInfos should not be empty")
+
+ // Verify structure for each operator
+ operatorIdToCkptInfos.foreach { case (operatorId, partitionToCkptIds) =>
+ // Should have checkpoint IDs for all partitions
+ assert(partitionToCkptIds.length == expectedShufflePartitions,
+ s"Operator $operatorId: Expected $expectedShufflePartitions partition checkpoint IDs, " +
+ s"but found ${partitionToCkptIds.length}")
+ // Each partition should have checkpoint IDs (at least one per store)
+ partitionToCkptIds.foreach { ckptIds =>
+ assert(ckptIds.nonEmpty, s"Operator $operatorId should have checkpoint IDs")
+ // Each checkpoint ID should be a non-empty string
+ ckptIds.foreach { ckptId =>
Review Comment:
We also need to assert that the number of ckptIds for the partition, matches the number of state stores the operator has. You can read the operator metadata to know how many stores per operatorId.
##########
sql/core/src/test/scala/org/apache/spark/sql/execution/streaming/state/RocksDBSuite.scala:
##########
@@ -3943,7 +3943,7 @@ class RocksDBSuite extends AlsoTestWithRocksDBFeatures with SharedSparkSession
}}
}
- test("SPARK-54420: load with createEmpty creates empty store") {
+ testWithStateStoreCheckpointIds("SPARK-54420: load with createEmpty creates empty store") { _ =>
Review Comment:
This test case is actually not truly testing checkpointId, since you are not passing in the checkpointId you got from `db.commit` into `db.load`
##########
sql/core/src/test/scala/org/apache/spark/sql/execution/streaming/state/OfflineStateRepartitionSuite.scala:
##########
@@ -454,4 +470,40 @@ object OfflineStateRepartitionTestUtils {
}
}
}
+
+ def verifyCheckpointIds(
+ repartitionBatchId: Long,
+ checkpointMetadata: StreamingQueryCheckpointMetadata,
+ expectedShufflePartitions: Int): Unit = {
+ // Verify commit log has the repartition batch with checkpoint IDs
+ val commitOpt = checkpointMetadata.commitLog.get(repartitionBatchId)
+ assert(commitOpt.isDefined, s"Commit for batch $repartitionBatchId should exist")
+
+ val commitMetadata = commitOpt.get
+
+ // Verify stateUniqueIds is present for checkpoint V2
+ assert(commitMetadata.stateUniqueIds.isDefined,
+ "stateUniqueIds should be present in commit metadata when checkpoint version >= 2")
+
+ val operatorIdToCkptInfos = commitMetadata.stateUniqueIds.get
+ assert(operatorIdToCkptInfos.nonEmpty,
+ "operatorIdToCkptInfos should not be empty")
+
+ // Verify structure for each operator
+ operatorIdToCkptInfos.foreach { case (operatorId, partitionToCkptIds) =>
+ // Should have checkpoint IDs for all partitions
+ assert(partitionToCkptIds.length == expectedShufflePartitions,
+ s"Operator $operatorId: Expected $expectedShufflePartitions partition checkpoint IDs, " +
+ s"but found ${partitionToCkptIds.length}")
+ // Each partition should have checkpoint IDs (at least one per store)
+ partitionToCkptIds.foreach { ckptIds =>
+ assert(ckptIds.nonEmpty, s"Operator $operatorId should have checkpoint IDs")
+ // Each checkpoint ID should be a non-empty string
+ ckptIds.foreach { ckptId =>
+ assert(ckptId.nonEmpty,
+ s"Operator $operatorId: Checkpoint ID should be non-empty")
Review Comment:
ditto
##########
sql/core/src/test/scala/org/apache/spark/sql/execution/streaming/state/OfflineStateRepartitionSuite.scala:
##########
@@ -454,4 +470,40 @@ object OfflineStateRepartitionTestUtils {
}
}
}
+
+ def verifyCheckpointIds(
Review Comment:
private
##########
sql/core/src/test/scala/org/apache/spark/sql/execution/streaming/state/StatePartitionAllColumnFamiliesWriterSuite.scala:
##########
@@ -450,8 +464,28 @@ class StatePartitionAllColumnFamiliesWriterSuite extends StateDataSourceTestBase
}
}
- testWithChangelogConfig("SPARK-54420: aggregation state ver 1") {
- testRoundTripForAggrStateVersion(1)
+ // Run transformWithState tests with enable/disable checkpoint V2
+ Seq(1, 2).foreach { ckptVersion =>
+ def testWithChangelogAndCheckpointId(testName: String)(testFun: => Unit): Unit = {
+ test(s"$testName ($changelogCpTestSuffix, enableCkptId = ${ckptVersion >= 2})") {
+ withSQLConf(
+ "spark.sql.streaming.stateStore.rocksdb.changelogCheckpointing.enabled" ->
Review Comment:
why not only set the format version here, then call `testWithChangelogConfig`. Instead of duplicating it here
##########
sql/core/src/test/scala/org/apache/spark/sql/execution/streaming/state/OfflineStateRepartitionSuite.scala:
##########
@@ -106,71 +106,87 @@ class OfflineStateRepartitionSuite extends StreamTest
)
}
- test("Repartition: success, failure, retry") {
- withTempDir { dir =>
- val originalPartitions = 3
- val input = MemoryStream[Int]
- val batchId = runSimpleStreamQuery(originalPartitions, dir.getAbsolutePath, input)
- val checkpointMetadata = new StreamingQueryCheckpointMetadata(spark, dir.getAbsolutePath)
- // Shouldn't be seen as a repartition batch
- assert(!isRepartitionBatch(batchId, checkpointMetadata.offsetLog, dir.getAbsolutePath))
-
- // Trying to repartition to the same number should fail
- val ex = intercept[StateRepartitionShufflePartitionsAlreadyMatchError] {
- spark.streamingCheckpointManager.repartition(dir.getAbsolutePath, originalPartitions)
+ Seq(1, 2).foreach { ckptVersion =>
+ def testWithChangelogAndCheckpointId(testName: String)(testFun: => Unit): Unit = {
+ test(s"$testName (enableCkptId = ${ckptVersion >= 2})") {
+ withSQLConf(
+ SQLConf.STATE_STORE_CHECKPOINT_FORMAT_VERSION.key -> ckptVersion.toString) {
+ testFun
+ }
}
- checkError(
- ex,
- condition = "STATE_REPARTITION_INVALID_CHECKPOINT.SHUFFLE_PARTITIONS_ALREADY_MATCH",
- parameters = Map(
- "checkpointLocation" -> dir.getAbsolutePath,
- "batchId" -> batchId.toString,
- "numPartitions" -> originalPartitions.toString
- )
- )
-
- // Trying to repartition to a different number should succeed
- val newPartitions = originalPartitions + 1
- spark.streamingCheckpointManager.repartition(dir.getAbsolutePath, newPartitions)
- val repartitionBatchId = batchId + 1
- val hadoopConf = spark.sessionState.newHadoopConf()
- verifyRepartitionBatch(
- repartitionBatchId, checkpointMetadata, hadoopConf, dir.getAbsolutePath, newPartitions)
-
- // Now delete the repartition commit to simulate a failed repartition attempt.
- // This will delete all the commits after the batchId.
- checkpointMetadata.commitLog.purgeAfter(batchId)
+ }
- // Try to repartition with a different numPartitions should fail,
- // since it will see an uncommitted repartition batch with a different numPartitions.
- val ex2 = intercept[StateRepartitionLastBatchAbandonedRepartitionError] {
- spark.streamingCheckpointManager.repartition(dir.getAbsolutePath, newPartitions + 1)
- }
- checkError(
- ex2,
- condition = "STATE_REPARTITION_INVALID_CHECKPOINT.LAST_BATCH_ABANDONED_REPARTITION",
- parameters = Map(
- "checkpointLocation" -> dir.getAbsolutePath,
- "lastBatchId" -> repartitionBatchId.toString,
- "lastBatchShufflePartitions" -> newPartitions.toString,
- "numPartitions" -> (newPartitions + 1).toString
+ testWithChangelogAndCheckpointId("Repartition: success, failure, retry") {
+ withTempDir { dir =>
+ val originalPartitions = 3
+ val input = MemoryStream[Int]
+ val batchId = runSimpleStreamQuery(originalPartitions, dir.getAbsolutePath, input)
+ val checkpointMetadata = new StreamingQueryCheckpointMetadata(spark, dir.getAbsolutePath)
+ // Shouldn't be seen as a repartition batch
+ assert(!isRepartitionBatch(batchId, checkpointMetadata.offsetLog, dir.getAbsolutePath))
+
+ // Trying to repartition to the same number should fail
+ val ex = intercept[StateRepartitionShufflePartitionsAlreadyMatchError] {
+ spark.streamingCheckpointManager.repartition(dir.getAbsolutePath, originalPartitions)
+ }
+ checkError(
+ ex,
+ condition = "STATE_REPARTITION_INVALID_CHECKPOINT.SHUFFLE_PARTITIONS_ALREADY_MATCH",
+ parameters = Map(
+ "checkpointLocation" -> dir.getAbsolutePath,
+ "batchId" -> batchId.toString,
+ "numPartitions" -> originalPartitions.toString
+ )
)
- )
- // Retrying with the same numPartitions should work
- spark.streamingCheckpointManager.repartition(dir.getAbsolutePath, newPartitions)
- verifyRepartitionBatch(
- repartitionBatchId, checkpointMetadata, hadoopConf, dir.getAbsolutePath, newPartitions)
+ // Trying to repartition to a different number should succeed
+ val newPartitions = originalPartitions + 1
+ spark.streamingCheckpointManager.repartition(dir.getAbsolutePath, newPartitions)
+ val repartitionBatchId = batchId + 1
+ val hadoopConf = spark.sessionState.newHadoopConf()
+ verifyRepartitionBatch(
+ repartitionBatchId, checkpointMetadata, hadoopConf, dir.getAbsolutePath, newPartitions)
+
+ // Now delete the repartition commit to simulate a failed repartition attempt.
+ // This will delete all the commits after the batchId.
+ checkpointMetadata.commitLog.purgeAfter(batchId)
+
+ // Try to repartition with a different numPartitions should fail,
+ // since it will see an uncommitted repartition batch with a different numPartitions.
+ val ex2 = intercept[StateRepartitionLastBatchAbandonedRepartitionError] {
+ spark.streamingCheckpointManager.repartition(dir.getAbsolutePath, newPartitions + 1)
+ }
+ checkError(
+ ex2,
+ condition = "STATE_REPARTITION_INVALID_CHECKPOINT.LAST_BATCH_ABANDONED_REPARTITION",
+ parameters = Map(
+ "checkpointLocation" -> dir.getAbsolutePath,
+ "lastBatchId" -> repartitionBatchId.toString,
+ "lastBatchShufflePartitions" -> newPartitions.toString,
+ "numPartitions" -> (newPartitions + 1).toString
+ )
+ )
- // Repartition with way more partitions, to verify that empty partitions are properly created
- val morePartitions = newPartitions * 3
- spark.streamingCheckpointManager.repartition(dir.getAbsolutePath, morePartitions)
- verifyRepartitionBatch(
- repartitionBatchId + 1, checkpointMetadata, hadoopConf,
- dir.getAbsolutePath, morePartitions)
+ // Retrying with the same numPartitions should work
+ spark.streamingCheckpointManager.repartition(dir.getAbsolutePath, newPartitions)
+ verifyRepartitionBatch(
+ repartitionBatchId, checkpointMetadata, hadoopConf, dir.getAbsolutePath, newPartitions)
+
+ // Repartition with way more partitions, to verify that empty partitions are properly
+ // created
+ val morePartitions = newPartitions * 3
+ spark.streamingCheckpointManager.repartition(dir.getAbsolutePath, morePartitions)
+ verifyRepartitionBatch(
+ repartitionBatchId + 1, checkpointMetadata, hadoopConf,
+ dir.getAbsolutePath, morePartitions)
+ if (spark.conf.get(
+ SQLConf.STATE_STORE_CHECKPOINT_FORMAT_VERSION.key).toInt >= 2) {
+ verifyCheckpointIds(repartitionBatchId + 1, checkpointMetadata, morePartitions)
Review Comment:
Lets call this `verifyCheckpointIds` within `verifyRepartitionBatch`, just like we do for the other verifications. So that by calling `verifyRepartitionBatch`, it does all the necessary verifications.
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/state/OfflineStateRepartitionRunner.scala:
##########
@@ -289,12 +292,32 @@ class OfflineStateRepartitionRunner(
}
}
- private def commitBatch(newBatchId: Long, lastCommittedBatchId: Long): Unit = {
+ private def commitBatch(
+ newBatchId: Long,
+ lastCommittedBatchId: Long,
+ opIdToStateStoreCkptInfo: Option[Map[Long, Array[Array[StateStoreCheckpointInfo]]]]): Unit = {
val latestCommit = checkpointMetadata.commitLog.get(lastCommittedBatchId).get
-
- // todo: For checkpoint v2, we need to update the stateUniqueIds based on the
- // newly created state commit. Will be done in subsequent PR.
- if (!checkpointMetadata.commitLog.add(newBatchId, latestCommit)) {
+ val commitMetadata = opIdToStateStoreCkptInfo.map {originalInfoMap =>
+ // opIdToStateStoreCkptInfo is Map[operatorId, Array[stateStore -> Array[partition -> info]]]
+ // we change it to Map[operatorId, Array[partitionId -> Array[store -> info]]]
+ val opIdToPartitionCkptInfo: Map[Long, Array[Array[String]]] =
+ originalInfoMap.map {
+ case(operator, storesSeq) =>
+ val numPartitions = storesSeq.head.length
+ operator -> (0 until numPartitions).map { partitionIdx =>
+ storesSeq.flatMap { storePartitions =>
+ storePartitions(partitionIdx).stateStoreCkptId
Review Comment:
When we move this conversion to rewriter. We should add some assertions i.e.
1. that the `storePartitions(partitionIdx).partitionId == partitionIdx`
2. that the `storePartitions(partitionIdx).batchVersion` is what we expect
3. same for `storePartitions(partitionIdx).baseStateStoreCkptId == None`
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-54590][SS] Support CheckpointV2 for StatePartitionAllColumnFamiliesWriter [spark]
Posted by "zifeif2 (via GitHub)" <gi...@apache.org>.
zifeif2 commented on code in PR #53720:
URL: https://github.com/apache/spark/pull/53720#discussion_r2670551584
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/state/RocksDB.scala:
##########
@@ -443,77 +443,82 @@ class RocksDB(
private def loadWithCheckpointId(
version: Long,
stateStoreCkptId: Option[String],
- readOnly: Boolean = false): RocksDB = {
+ readOnly: Boolean = false,
+ loadEmpty: Boolean = false): RocksDB = {
// An array contains lineage information from [snapShotVersion, version]
// (inclusive in both ends)
var currVersionLineage: Array[LineageItem] = lineageManager.getLineageForCurrVersion()
try {
- if (loadedVersion != version || (loadedStateStoreCkptId.isEmpty ||
- stateStoreCkptId.get != loadedStateStoreCkptId.get)) {
+ if (loadEmpty || loadedVersion != version || loadedStateStoreCkptId.isEmpty ||
+ stateStoreCkptId.get != loadedStateStoreCkptId.get) {
closeDB(ignoreException = false)
-
- val (latestSnapshotVersion, latestSnapshotUniqueId) = {
- // Special handling when version is 0.
- // When loading the very first version (0), stateStoreCkptId does not need to be defined
- // because there won't be 0.changelog / 0.zip file created in RocksDB under v2.
- if (version == 0) {
- assert(stateStoreCkptId.isEmpty,
- "stateStoreCkptId should be empty when version is zero")
- (0L, None)
- // When there is a snapshot file, it is the ground truth, we can skip
- // reconstructing the lineage from changelog file.
- } else if (fileManager.existsSnapshotFile(version, stateStoreCkptId)) {
- currVersionLineage = Array(LineageItem(version, stateStoreCkptId.get))
- (version, stateStoreCkptId)
- } else {
- currVersionLineage = getLineageFromChangelogFile(version, stateStoreCkptId) :+
- LineageItem(version, stateStoreCkptId.get)
- currVersionLineage = currVersionLineage.sortBy(_.version)
-
- val latestSnapshotVersionsAndUniqueId =
- fileManager.getLatestSnapshotVersionAndUniqueIdFromLineage(currVersionLineage)
- latestSnapshotVersionsAndUniqueId match {
- case Some(pair) => (pair._1, Option(pair._2))
- case None if currVersionLineage.head.version == 1L =>
- logDebug(log"Cannot find latest snapshot based on lineage but first version " +
- log"is 1, use 0 as default. Lineage: ${MDC(LogKeys.LINEAGE, lineageManager)}")
- (0L, None)
- case _ =>
- throw QueryExecutionErrors.cannotFindBaseSnapshotCheckpoint(
- printLineageItems(currVersionLineage))
+ if (loadEmpty) {
+ loadEmptyStore(version)
+ lineageManager.clear()
+ } else {
+ val (latestSnapshotVersion, latestSnapshotUniqueId) = {
+ // Special handling when version is 0.
+ // When loading the very first version (0), stateStoreCkptId does not need to be defined
+ // because there won't be 0.changelog / 0.zip file created in RocksDB under v2.
+ if (version == 0) {
+ assert(stateStoreCkptId.isEmpty,
+ "stateStoreCkptId should be empty when version is zero")
+ (0L, None)
+ // When there is a snapshot file, it is the ground truth, we can skip
+ // reconstructing the lineage from changelog file.
+ } else if (fileManager.existsSnapshotFile(version, stateStoreCkptId)) {
+ currVersionLineage = Array(LineageItem(version, stateStoreCkptId.get))
+ (version, stateStoreCkptId)
+ } else {
+ currVersionLineage = getLineageFromChangelogFile(version, stateStoreCkptId) :+
+ LineageItem(version, stateStoreCkptId.get)
+ currVersionLineage = currVersionLineage.sortBy(_.version)
+
+ val latestSnapshotVersionsAndUniqueId =
+ fileManager.getLatestSnapshotVersionAndUniqueIdFromLineage(currVersionLineage)
+ latestSnapshotVersionsAndUniqueId match {
+ case Some(pair) => (pair._1, Option(pair._2))
+ case None if currVersionLineage.head.version == 1L =>
+ logDebug(log"Cannot find latest snapshot based on lineage but first version " +
+ log"is 1, use 0 as default. Lineage: ${MDC(LogKeys.LINEAGE, lineageManager)}")
+ (0L, None)
+ case _ =>
+ throw QueryExecutionErrors.cannotFindBaseSnapshotCheckpoint(
+ printLineageItems(currVersionLineage))
+ }
}
}
- }
- logInfo(log"Loaded latestSnapshotVersion: ${
- MDC(LogKeys.SNAPSHOT_VERSION, latestSnapshotVersion)}, latestSnapshotUniqueId: ${
- MDC(LogKeys.UUID, latestSnapshotUniqueId)}")
+ logInfo(log"Loaded latestSnapshotVersion: ${
+ MDC(LogKeys.SNAPSHOT_VERSION, latestSnapshotVersion)}, latestSnapshotUniqueId: ${
+ MDC(LogKeys.UUID, latestSnapshotUniqueId)}")
- val metadata = fileManager.loadCheckpointFromDfs(latestSnapshotVersion,
- workingDir, rocksDBFileMapping, latestSnapshotUniqueId)
+ val metadata = fileManager.loadCheckpointFromDfs(latestSnapshotVersion,
+ workingDir, rocksDBFileMapping, latestSnapshotUniqueId)
- loadedVersion = latestSnapshotVersion
+ loadedVersion = latestSnapshotVersion
- // reset the last snapshot version to the latest available snapshot version
- lastSnapshotVersion = latestSnapshotVersion
- lineageManager.resetLineage(currVersionLineage)
+ // reset the last snapshot version to the latest available snapshot version
+ lastSnapshotVersion = latestSnapshotVersion
+ lineageManager.resetLineage(currVersionLineage)
- // Initialize maxVersion upon successful load from DFS
- fileManager.setMaxSeenVersion(version)
+ // Initialize maxVersion upon successful load from DFS
+ fileManager.setMaxSeenVersion(version)
- // Report this snapshot version to the coordinator
- reportSnapshotUploadToCoordinator(latestSnapshotVersion)
+ // Report this snapshot version to the coordinator
+ reportSnapshotUploadToCoordinator(latestSnapshotVersion)
- openLocalRocksDB(metadata)
+ openLocalRocksDB(metadata)
- if (loadedVersion != version) {
- val versionsAndUniqueIds = currVersionLineage.collect {
+ if (loadedVersion != version) {
+ val versionsAndUniqueIds = currVersionLineage.collect {
case i if i.version > loadedVersion && i.version <= version =>
(i.version, Option(i.checkpointUniqueId))
}
- replayChangelog(versionsAndUniqueIds)
- loadedVersion = version
- lineageManager.resetLineage(currVersionLineage)
+ replayChangelog(versionsAndUniqueIds)
+ loadedVersion = version
+ lineageManager.resetLineage(currVersionLineage)
+ }
Review Comment:
These lines are the same as old version except for indentation
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-54590][SS] Support Checkpoint V2 for State Rewriter and Repartitioning [spark]
Posted by "anishshri-db (via GitHub)" <gi...@apache.org>.
anishshri-db commented on code in PR #53720:
URL: https://github.com/apache/spark/pull/53720#discussion_r2696008570
##########
common/utils/src/main/resources/error/error-conditions.json:
##########
@@ -5603,6 +5603,13 @@
"Ensure that the checkpoint is for a stateful streaming query and the query ran on a Spark version that supports operator metadata (Spark 4.0+)."
]
},
+ "STATE_CHECKPOINT_FORMAT_VERSION_MISMATCH" : {
+ "message" : [
+ "The checkpoint format version in SQLConf does not match the checkpoint version in the commit log.",
+ "Expected version v<expectedVersion>, but found v<actualVersion>.",
Review Comment:
typo here ?
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-54590][SS] Support Checkpoint V2 for State Rewriter and Repartitioning [spark]
Posted by "zifeif2 (via GitHub)" <gi...@apache.org>.
zifeif2 commented on code in PR #53720:
URL: https://github.com/apache/spark/pull/53720#discussion_r2696221821
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/state/RocksDB.scala:
##########
@@ -443,77 +443,84 @@ class RocksDB(
private def loadWithCheckpointId(
version: Long,
stateStoreCkptId: Option[String],
- readOnly: Boolean = false): RocksDB = {
+ readOnly: Boolean = false,
+ loadEmpty: Boolean = false): RocksDB = {
// An array contains lineage information from [snapShotVersion, version]
// (inclusive in both ends)
var currVersionLineage: Array[LineageItem] = lineageManager.getLineageForCurrVersion()
try {
- if (loadedVersion != version || (loadedStateStoreCkptId.isEmpty ||
- stateStoreCkptId.get != loadedStateStoreCkptId.get)) {
+ if (loadEmpty || loadedVersion != version || loadedStateStoreCkptId.isEmpty ||
+ stateStoreCkptId.get != loadedStateStoreCkptId.get) {
closeDB(ignoreException = false)
-
- val (latestSnapshotVersion, latestSnapshotUniqueId) = {
- // Special handling when version is 0.
- // When loading the very first version (0), stateStoreCkptId does not need to be defined
- // because there won't be 0.changelog / 0.zip file created in RocksDB under v2.
- if (version == 0) {
- assert(stateStoreCkptId.isEmpty,
- "stateStoreCkptId should be empty when version is zero")
- (0L, None)
- // When there is a snapshot file, it is the ground truth, we can skip
- // reconstructing the lineage from changelog file.
- } else if (fileManager.existsSnapshotFile(version, stateStoreCkptId)) {
- currVersionLineage = Array(LineageItem(version, stateStoreCkptId.get))
- (version, stateStoreCkptId)
- } else {
- currVersionLineage = getLineageFromChangelogFile(version, stateStoreCkptId) :+
- LineageItem(version, stateStoreCkptId.get)
- currVersionLineage = currVersionLineage.sortBy(_.version)
-
- val latestSnapshotVersionsAndUniqueId =
- fileManager.getLatestSnapshotVersionAndUniqueIdFromLineage(currVersionLineage)
- latestSnapshotVersionsAndUniqueId match {
- case Some(pair) => (pair._1, Option(pair._2))
- case None if currVersionLineage.head.version == 1L =>
- logDebug(log"Cannot find latest snapshot based on lineage but first version " +
- log"is 1, use 0 as default. Lineage: ${MDC(LogKeys.LINEAGE, lineageManager)}")
- (0L, None)
- case _ =>
- throw QueryExecutionErrors.cannotFindBaseSnapshotCheckpoint(
- printLineageItems(currVersionLineage))
+ if (loadEmpty) {
+ require(stateStoreCkptId.isEmpty,
+ "stateStoreCkptId should be empty when loadEmpty is true")
+ loadEmptyStore(version)
+ lineageManager.clear()
+ } else {
+ val (latestSnapshotVersion, latestSnapshotUniqueId) = {
+ // Special handling when version is 0.
+ // When loading the very first version (0), stateStoreCkptId does not need to be defined
+ // because there won't be 0.changelog / 0.zip file created in RocksDB under v2.
+ if (version == 0) {
+ assert(stateStoreCkptId.isEmpty,
+ "stateStoreCkptId should be empty when version is zero")
+ (0L, None)
+ // When there is a snapshot file, it is the ground truth, we can skip
+ // reconstructing the lineage from changelog file.
+ } else if (fileManager.existsSnapshotFile(version, stateStoreCkptId)) {
+ currVersionLineage = Array(LineageItem(version, stateStoreCkptId.get))
+ (version, stateStoreCkptId)
+ } else {
+ currVersionLineage = getLineageFromChangelogFile(version, stateStoreCkptId) :+
+ LineageItem(version, stateStoreCkptId.get)
+ currVersionLineage = currVersionLineage.sortBy(_.version)
+
+ val latestSnapshotVersionsAndUniqueId =
+ fileManager.getLatestSnapshotVersionAndUniqueIdFromLineage(currVersionLineage)
+ latestSnapshotVersionsAndUniqueId match {
+ case Some(pair) => (pair._1, Option(pair._2))
+ case None if currVersionLineage.head.version == 1L =>
+ logDebug(log"Cannot find latest snapshot based on lineage but first version " +
+ log"is 1, use 0 as default. Lineage: ${MDC(LogKeys.LINEAGE, lineageManager)}")
+ (0L, None)
+ case _ =>
+ throw QueryExecutionErrors.cannotFindBaseSnapshotCheckpoint(
+ printLineageItems(currVersionLineage))
+ }
}
}
- }
- logInfo(log"Loaded latestSnapshotVersion: ${
- MDC(LogKeys.SNAPSHOT_VERSION, latestSnapshotVersion)}, latestSnapshotUniqueId: ${
- MDC(LogKeys.UUID, latestSnapshotUniqueId)}")
+ logInfo(log"Loaded latestSnapshotVersion: ${
Review Comment:
Refactored the if/else block and unindented it
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-54590][SS] Support CheckpointV2 for StatePartitionAllColumnFamiliesWriter [spark]
Posted by "github-actions[bot] (via GitHub)" <gi...@apache.org>.
github-actions[bot] commented on PR #53720:
URL: https://github.com/apache/spark/pull/53720#issuecomment-3721383972
## JIRA Issue Information
=== New Feature SPARK-54590 ===
Summary: State Writer supports checkpoint V2
Assignee: None
Status: Open
Affected: ["4.1.0"]
---
*This comment was automatically generated by GitHub Actions*
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-54590][SS] Support CheckpointV2 for StatePartitionAllColumnFamiliesWriter [spark]
Posted by "zifeif2 (via GitHub)" <gi...@apache.org>.
zifeif2 commented on code in PR #53720:
URL: https://github.com/apache/spark/pull/53720#discussion_r2695390618
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/state/StateRewriter.scala:
##########
@@ -224,9 +261,8 @@ class StateRewriter(
schemaProvider,
executorSqlConf
)
-
- partitionWriter.write(partitionIter)
- }
+ Iterator(partitionWriter.write(partitionIter))
+ }.sortBy(_.partitionId).collect()
Review Comment:
Actually yes, `collect` will sort the partition by Id. I was printing partitionId within the map function and saw that they are not ordered, but TIL `collect` will sort the result afterwards.
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-54590][SS] Support Checkpoint V2 for State Rewriter and Repartitioning [spark]
Posted by "zifeif2 (via GitHub)" <gi...@apache.org>.
zifeif2 commented on code in PR #53720:
URL: https://github.com/apache/spark/pull/53720#discussion_r2696211902
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/state/RocksDBFileManager.scala:
##########
@@ -375,6 +375,8 @@ class RocksDBFileManager(
createDfsRootDirIfNotExist()
// Since we cleared the local dir, we should also clear the local file mapping
rocksDBFileMapping.clear()
+ // Set empty metrics since we're not loading any files from DFS
+ loadCheckpointMetrics = RocksDBFileManagerMetrics.EMPTY_METRICS
Review Comment:
Added this line to make sure that loadCheckpointMetrics is set correctly when we are loading empty store cc @anishshri-db .
RocksDB will run `fileManagerMetrics = fileManager.latestLoadCheckpointMetrics`, and `latestLoadCheckpointMetrics` return `loadCheckpointMetrics`
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-54590][SS] Support Checkpoint V2 for State Rewriter and Repartitioning [spark]
Posted by "zifeif2 (via GitHub)" <gi...@apache.org>.
zifeif2 commented on code in PR #53720:
URL: https://github.com/apache/spark/pull/53720#discussion_r2696211902
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/state/RocksDBFileManager.scala:
##########
@@ -375,6 +375,8 @@ class RocksDBFileManager(
createDfsRootDirIfNotExist()
// Since we cleared the local dir, we should also clear the local file mapping
rocksDBFileMapping.clear()
+ // Set empty metrics since we're not loading any files from DFS
+ loadCheckpointMetrics = RocksDBFileManagerMetrics.EMPTY_METRICS
Review Comment:
Added this line to make sure that loadCheckpointMetrics is set correctly when we are loading empty store
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-54590][SS] Support CheckpointV2 for StatePartitionAllColumnFamiliesWriter [spark]
Posted by "micheal-o (via GitHub)" <gi...@apache.org>.
micheal-o commented on code in PR #53720:
URL: https://github.com/apache/spark/pull/53720#discussion_r2687692566
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/state/OfflineStateRepartitionRunner.scala:
##########
@@ -289,12 +290,27 @@ class OfflineStateRepartitionRunner(
}
}
- private def commitBatch(newBatchId: Long, lastCommittedBatchId: Long): Unit = {
+ private def commitBatch(
+ newBatchId: Long,
+ lastCommittedBatchId: Long,
+ opIdToStateStoreCkptInfo: Map[Long, Array[Array[StateStoreCheckpointInfo]]]): Unit = {
+ val enableCheckpointId = StatefulOperatorStateInfo.
+ enableStateStoreCheckpointIds(sparkSession.sessionState.conf)
val latestCommit = checkpointMetadata.commitLog.get(lastCommittedBatchId).get
+ val commitMetadata = if (enableCheckpointId) {
+ val opIdToPartitionCkptInfo: Map[Long, Array[Array[String]]] =
+ opIdToStateStoreCkptInfo.map {
+ case (operator, partitionSeq) =>
+ operator -> partitionSeq.map(storeSeq =>
+ storeSeq.flatMap(info => info.stateStoreCkptId))
Review Comment:
Why are we still doing this conversion here. I mentioned this previously that it should be done in StateRewriter, to avoid having all callers implementing this additional conversion
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/state/OfflineStateRepartitionRunner.scala:
##########
@@ -289,12 +290,27 @@ class OfflineStateRepartitionRunner(
}
}
- private def commitBatch(newBatchId: Long, lastCommittedBatchId: Long): Unit = {
+ private def commitBatch(
+ newBatchId: Long,
+ lastCommittedBatchId: Long,
+ opIdToStateStoreCkptInfo: Map[Long, Array[Array[StateStoreCheckpointInfo]]]): Unit = {
+ val enableCheckpointId = StatefulOperatorStateInfo.
+ enableStateStoreCheckpointIds(sparkSession.sessionState.conf)
val latestCommit = checkpointMetadata.commitLog.get(lastCommittedBatchId).get
+ val commitMetadata = if (enableCheckpointId) {
Review Comment:
Why are we doing this check? Why not just check if checkpointId was passed in?
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/state/RocksDB.scala:
##########
@@ -443,77 +443,84 @@ class RocksDB(
private def loadWithCheckpointId(
version: Long,
stateStoreCkptId: Option[String],
- readOnly: Boolean = false): RocksDB = {
+ readOnly: Boolean = false,
+ loadEmpty: Boolean = false): RocksDB = {
// An array contains lineage information from [snapShotVersion, version]
// (inclusive in both ends)
var currVersionLineage: Array[LineageItem] = lineageManager.getLineageForCurrVersion()
try {
- if (loadedVersion != version || (loadedStateStoreCkptId.isEmpty ||
- stateStoreCkptId.get != loadedStateStoreCkptId.get)) {
+ if (loadEmpty || loadedVersion != version || loadedStateStoreCkptId.isEmpty ||
+ stateStoreCkptId.get != loadedStateStoreCkptId.get) {
closeDB(ignoreException = false)
-
- val (latestSnapshotVersion, latestSnapshotUniqueId) = {
- // Special handling when version is 0.
- // When loading the very first version (0), stateStoreCkptId does not need to be defined
- // because there won't be 0.changelog / 0.zip file created in RocksDB under v2.
- if (version == 0) {
- assert(stateStoreCkptId.isEmpty,
- "stateStoreCkptId should be empty when version is zero")
- (0L, None)
- // When there is a snapshot file, it is the ground truth, we can skip
- // reconstructing the lineage from changelog file.
- } else if (fileManager.existsSnapshotFile(version, stateStoreCkptId)) {
- currVersionLineage = Array(LineageItem(version, stateStoreCkptId.get))
- (version, stateStoreCkptId)
- } else {
- currVersionLineage = getLineageFromChangelogFile(version, stateStoreCkptId) :+
- LineageItem(version, stateStoreCkptId.get)
- currVersionLineage = currVersionLineage.sortBy(_.version)
-
- val latestSnapshotVersionsAndUniqueId =
- fileManager.getLatestSnapshotVersionAndUniqueIdFromLineage(currVersionLineage)
- latestSnapshotVersionsAndUniqueId match {
- case Some(pair) => (pair._1, Option(pair._2))
- case None if currVersionLineage.head.version == 1L =>
- logDebug(log"Cannot find latest snapshot based on lineage but first version " +
- log"is 1, use 0 as default. Lineage: ${MDC(LogKeys.LINEAGE, lineageManager)}")
- (0L, None)
- case _ =>
- throw QueryExecutionErrors.cannotFindBaseSnapshotCheckpoint(
- printLineageItems(currVersionLineage))
+ if (loadEmpty) {
+ require(stateStoreCkptId.isEmpty,
+ "stateStoeCkptId should be empty when loadEmpty is true")
Review Comment:
nit: typo `stateStoreCkptId`
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/state/RocksDB.scala:
##########
@@ -530,9 +537,13 @@ class RocksDB(
if (conf.resetStatsOnLoad) {
nativeStats.reset
}
-
- logInfo(log"Loaded ${MDC(LogKeys.VERSION_NUM, version)} " +
- log"with uniqueId ${MDC(LogKeys.UUID, stateStoreCkptId)}")
+ if (loadEmpty) {
+ logInfo(log"Loaded empty store at version ${MDC(LogKeys.VERSION_NUM, version)} " +
+ log"without uniqueId")
Review Comment:
nit: This can be confusing and someone might think this is for loadWithoutCheckpointId. Just say "with uniqueId"
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/state/StateRewriter.scala:
##########
@@ -81,24 +82,46 @@ class StateRewriter(
private val stateRootLocation = new Path(
resolvedCheckpointLocation, StreamingCheckpointConstants.DIR_NAME_STATE).toString
- def run(): Unit = {
+ // return a Map[operator id, Array[partition -> Array[stateStore -> StateStoreCheckpointInfo]]]
+ def run(): Map[Long, Array[Array[StateStoreCheckpointInfo]]] = {
logInfo(log"Starting state rewrite for " +
log"checkpointLocation=${MDC(CHECKPOINT_LOCATION, resolvedCheckpointLocation)}, " +
log"readCheckpointLocation=" +
log"${MDC(CHECKPOINT_LOCATION, readResolvedCheckpointLocation.getOrElse(""))}, " +
log"readBatchId=${MDC(BATCH_ID, readBatchId)}, " +
log"writeBatchId=${MDC(BATCH_ID, writeBatchId)}")
- val (_, timeTakenMs) = Utils.timeTakenMs {
+ val (checkpointInfos, timeTakenMs) = Utils.timeTakenMs {
runInternal()
}
+ val allCheckpointInfos = checkpointInfos.values.flatMap { checkpointInfos =>
+ checkpointInfos.flatten.toSeq
+ }.toSeq
logInfo(log"State rewrite completed in ${MDC(DURATION, timeTakenMs)} ms for " +
- log"checkpointLocation=${MDC(CHECKPOINT_LOCATION, resolvedCheckpointLocation)}")
+ log"checkpointLocation=${MDC(CHECKPOINT_LOCATION, resolvedCheckpointLocation)} " +
+ log"checkpointInfos=${MDC(LAST_COMMITTED_CHECKPOINT_ID, allCheckpointInfos)}"
+ )
+ checkpointInfos.map {
Review Comment:
this should be a separate func. Then it should be called within `runInternal`
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/state/StateRewriter.scala:
##########
@@ -81,24 +82,46 @@ class StateRewriter(
private val stateRootLocation = new Path(
resolvedCheckpointLocation, StreamingCheckpointConstants.DIR_NAME_STATE).toString
- def run(): Unit = {
+ // return a Map[operator id, Array[partition -> Array[stateStore -> StateStoreCheckpointInfo]]]
+ def run(): Map[Long, Array[Array[StateStoreCheckpointInfo]]] = {
Review Comment:
Also, should return the structure needed in commit Log. So callers don't need to do extra conversion work.
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/state/StateRewriter.scala:
##########
@@ -81,24 +82,46 @@ class StateRewriter(
private val stateRootLocation = new Path(
resolvedCheckpointLocation, StreamingCheckpointConstants.DIR_NAME_STATE).toString
- def run(): Unit = {
+ // return a Map[operator id, Array[partition -> Array[stateStore -> StateStoreCheckpointInfo]]]
+ def run(): Map[Long, Array[Array[StateStoreCheckpointInfo]]] = {
logInfo(log"Starting state rewrite for " +
log"checkpointLocation=${MDC(CHECKPOINT_LOCATION, resolvedCheckpointLocation)}, " +
log"readCheckpointLocation=" +
log"${MDC(CHECKPOINT_LOCATION, readResolvedCheckpointLocation.getOrElse(""))}, " +
log"readBatchId=${MDC(BATCH_ID, readBatchId)}, " +
log"writeBatchId=${MDC(BATCH_ID, writeBatchId)}")
- val (_, timeTakenMs) = Utils.timeTakenMs {
+ val (checkpointInfos, timeTakenMs) = Utils.timeTakenMs {
runInternal()
}
+ val allCheckpointInfos = checkpointInfos.values.flatMap { checkpointInfos =>
+ checkpointInfos.flatten.toSeq
+ }.toSeq
logInfo(log"State rewrite completed in ${MDC(DURATION, timeTakenMs)} ms for " +
- log"checkpointLocation=${MDC(CHECKPOINT_LOCATION, resolvedCheckpointLocation)}")
+ log"checkpointLocation=${MDC(CHECKPOINT_LOCATION, resolvedCheckpointLocation)} " +
+ log"checkpointInfos=${MDC(LAST_COMMITTED_CHECKPOINT_ID, allCheckpointInfos)}"
+ )
+ checkpointInfos.map {
Review Comment:
The `run` function here should just return the checkpointIds without doing any extra conversion work
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/state/StateRewriter.scala:
##########
@@ -81,24 +82,46 @@ class StateRewriter(
private val stateRootLocation = new Path(
resolvedCheckpointLocation, StreamingCheckpointConstants.DIR_NAME_STATE).toString
- def run(): Unit = {
+ // return a Map[operator id, Array[partition -> Array[stateStore -> StateStoreCheckpointInfo]]]
+ def run(): Map[Long, Array[Array[StateStoreCheckpointInfo]]] = {
logInfo(log"Starting state rewrite for " +
log"checkpointLocation=${MDC(CHECKPOINT_LOCATION, resolvedCheckpointLocation)}, " +
log"readCheckpointLocation=" +
log"${MDC(CHECKPOINT_LOCATION, readResolvedCheckpointLocation.getOrElse(""))}, " +
log"readBatchId=${MDC(BATCH_ID, readBatchId)}, " +
log"writeBatchId=${MDC(BATCH_ID, writeBatchId)}")
- val (_, timeTakenMs) = Utils.timeTakenMs {
+ val (checkpointInfos, timeTakenMs) = Utils.timeTakenMs {
runInternal()
}
+ val allCheckpointInfos = checkpointInfos.values.flatMap { checkpointInfos =>
+ checkpointInfos.flatten.toSeq
Review Comment:
Why are you doing this? Why not log the same thing as what you're returning. Makes it easier to debug.
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/state/StateRewriter.scala:
##########
@@ -81,24 +82,46 @@ class StateRewriter(
private val stateRootLocation = new Path(
resolvedCheckpointLocation, StreamingCheckpointConstants.DIR_NAME_STATE).toString
- def run(): Unit = {
+ // return a Map[operator id, Array[partition -> Array[stateStore -> StateStoreCheckpointInfo]]]
+ def run(): Map[Long, Array[Array[StateStoreCheckpointInfo]]] = {
Review Comment:
Should return an `Option`. Should return None if no checkpointIds
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/state/StateRewriter.scala:
##########
@@ -342,6 +366,21 @@ class StateRewriter(
None
}
}
+
+ private def setCheckpointVersion(): Unit = {
+ // Setting checkpoint version in sqlConf based on previous commitLog in case user forgot to
+ // set STATE_STORE_CHECKPOINT_FORMAT_VERSION and crash rewriter.
+ val commitLog = writeCheckpointMetadata.getOrElse(
+ new StreamingQueryCheckpointMetadata(sparkSession, resolvedCheckpointLocation)).commitLog
Review Comment:
I mentioned in my previous [comment](https://github.com/apache/spark/pull/53720#discussion_r2679166160) that we should be checking this for the `readBatchId`. The latest commit in the commitLog could be a skipped batch (e.g. skipped by repartition) with no checkpointId.
So we should do something like:
create 2 class vals:
```
private lazy val writeCheckpoint = writeCheckpointMetadata.getOrElse(
new StreamingQueryCheckpointMetadata(sparkSession, resolvedCheckpointLocation))
private lazy val readCheckpoint = if (readResolvedCheckpointLocation.isDefined) {
new StreamingQueryCheckpointMetadata(sparkSession, readResolvedCheckpointLocation.get)
} else {
// Same checkpoint for read & write
writeCheckpoint
}
```
Then in your func:
```
// Using read batch commit since the latest commit could be a skipped batch
val readBatchCommit = readCheckpoint.commitLog.get(readBatchId)
```
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/state/StateRewriter.scala:
##########
@@ -342,6 +366,21 @@ class StateRewriter(
None
}
}
+
+ private def setCheckpointVersion(): Unit = {
+ // Setting checkpoint version in sqlConf based on previous commitLog in case user forgot to
+ // set STATE_STORE_CHECKPOINT_FORMAT_VERSION and crash rewriter.
+ val commitLog = writeCheckpointMetadata.getOrElse(
+ new StreamingQueryCheckpointMetadata(sparkSession, resolvedCheckpointLocation)).commitLog
+ val latestCommitLog = commitLog.getLatest()
+ assert(latestCommitLog.isDefined)
+ if (latestCommitLog.get._2.stateUniqueIds.isDefined) {
+ // Hard code checkpoint version to 2 if previous commit log has checkpoint id.
+ // This prevents StateRewriter from crashing in case client forget to set correct
Review Comment:
This prevents the StateRewriter from failing to write the correct state files ...
##########
sql/core/src/test/scala/org/apache/spark/sql/execution/streaming/state/OfflineStateRepartitionSuite.scala:
##########
@@ -296,8 +323,21 @@ object OfflineStateRepartitionTestUtils {
verifyOffsetAndCommitLog(
batchId, previousBatchId, expectedShufflePartitions, checkpointMetadata)
verifyPartitionDirs(checkpointLocation, expectedShufflePartitions)
+
+ val serializableConf = new SerializableConfiguration(hadoopConf)
+ val baseOperatorsMetadata = getOperatorMetadata(
+ checkpointLocation, serializableConf, previousBatchId)
+ val repartitionOperatorsMetadata = getOperatorMetadata(
+ checkpointLocation, serializableConf, batchId)
verifyOperatorMetadata(
- batchId, previousBatchId, checkpointLocation, expectedShufflePartitions, hadoopConf)
+ baseOperatorsMetadata, repartitionOperatorsMetadata, expectedShufflePartitions)
+ if (StatefulOperatorStateInfo.enableStateStoreCheckpointIds(spark.sessionState.conf)) {
+ val expectedStoreCnt: Int = baseOperatorsMetadata.head match {
Review Comment:
move this into `verifyCheckpointIds`. Also, we need to get the store count per operator. Since this verification method can be called for test query with multiple operators.
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/state/StateRewriter.scala:
##########
@@ -342,6 +366,21 @@ class StateRewriter(
None
}
}
+
+ private def setCheckpointVersion(): Unit = {
+ // Setting checkpoint version in sqlConf based on previous commitLog in case user forgot to
+ // set STATE_STORE_CHECKPOINT_FORMAT_VERSION and crash rewriter.
+ val commitLog = writeCheckpointMetadata.getOrElse(
+ new StreamingQueryCheckpointMetadata(sparkSession, resolvedCheckpointLocation)).commitLog
+ val latestCommitLog = commitLog.getLatest()
+ assert(latestCommitLog.isDefined)
+ if (latestCommitLog.get._2.stateUniqueIds.isDefined) {
+ // Hard code checkpoint version to 2 if previous commit log has checkpoint id.
Review Comment:
"Set checkpoint..." and fix the comment
##########
sql/core/src/test/scala/org/apache/spark/sql/execution/streaming/state/OfflineStateRepartitionSuite.scala:
##########
@@ -376,23 +416,20 @@ object OfflineStateRepartitionTestUtils {
}
}
- private def verifyOperatorMetadata(
- repartitionBatchId: Long,
- baseBatchId: Long,
+ private def getOperatorMetadata(
checkpointLocation: String,
- expectedShufflePartitions: Int,
- hadoopConf: Configuration): Unit = {
- val serializableConf = new SerializableConfiguration(hadoopConf)
-
- // Read operator metadata for both batches
- val baseMetadataReader = new StateMetadataPartitionReader(
- checkpointLocation, serializableConf, baseBatchId)
+ serializableConf: SerializableConfiguration,
+ batchId: Long
+ ): Array[OperatorStateMetadata] = {
val repartitionMetadataReader = new StateMetadataPartitionReader(
Review Comment:
fix val name
##########
sql/core/src/test/scala/org/apache/spark/sql/execution/streaming/state/RocksDBSuite.scala:
##########
@@ -3943,47 +3943,55 @@ class RocksDBSuite extends AlsoTestWithRocksDBFeatures with SharedSparkSession
}}
}
- test("SPARK-54420: load with createEmpty creates empty store") {
- val remoteDir = Utils.createTempDir().toString
- new File(remoteDir).delete()
-
- withDB(remoteDir) { db =>
- // loading batch 0 with loadEmpty = true
- db.load(0, None, loadEmpty = true)
- assert(iterator(db).isEmpty)
- db.put("a", "1")
- val (version1, _) = db.commit()
- assert(toStr(db.get("a")) === "1")
+ testWithStateStoreCheckpointIds(
+ "SPARK-54420: load with createEmpty creates empty store") { enableCkptId =>
+ val remoteDir = Utils.createTempDir().toString
+ new File(remoteDir).delete()
- // check we can load store normally even the previous one loadEmpty = true
- db.load(version1)
- db.put("b", "2")
- val (version2, _) = db.commit()
- assert(version2 === version1 + 1)
- assert(toStr(db.get("b")) === "2")
- assert(toStr(db.get("a")) === "1")
+ withDB(remoteDir, enableStateStoreCheckpointIds = enableCkptId) { db =>
+ // loading batch 0 with loadEmpty = true
+ db.load(0, None, loadEmpty = true)
+ assert(iterator(db).isEmpty)
+ db.put("a", "1")
+ val (version1, checkpointInfoV1) = db.commit()
+ assert(toStr(db.get("a")) === "1")
- // load an empty store
- db.load(version2, loadEmpty = true)
- db.put("c", "3")
- val (version3, _) = db.commit()
- assert(db.get("b") === null)
- assert(db.get("a") === null)
- assert(toStr(db.get("c")) === "3")
- assert(version3 === version2 + 1)
-
- // load 2 empty store in a row
- db.load(version3, loadEmpty = true)
- db.put("d", "4")
- val (version4, _) = db.commit()
- assert(db.get("c") === null)
- assert(toStr(db.get("d")) === "4")
- assert(version4 === version3 + 1)
-
- db.load(version4)
- db.put("e", "5")
- db.commit()
- assert(db.iterator().map(toStr).toSet === Set(("d", "4"), ("e", "5")))
+ // check we can load store normally even the previous one loadEmpty = true
+ db.load(version1, checkpointInfoV1.stateStoreCkptId)
+ db.put("b", "2")
+ val (version2, _) = db.commit()
+ assert(version2 === version1 + 1)
+ assert(toStr(db.get("b")) === "2")
+ assert(toStr(db.get("a")) === "1")
+
+ // load an empty store
+ db.load(version2, loadEmpty = true)
+ db.put("c", "3")
+ val (version3, _) = db.commit()
+ assert(db.get("b") === null)
+ assert(db.get("a") === null)
+ assert(toStr(db.get("c")) === "3")
+ assert(version3 === version2 + 1)
+
+ // load 2 empty store in a row
+ db.load(version3, loadEmpty = true)
+ db.put("d", "4")
+ val (version4, checkpointInfoV4) = db.commit()
+ assert(db.get("c") === null)
+ assert(toStr(db.get("d")) === "4")
+ assert(version4 === version3 + 1)
+
+ db.load(version4, checkpointInfoV4.stateStoreCkptId)
Review Comment:
Use a new `withDB` for this one. To make sure we are reloading from checkpoint. Because currently, since last commit is v4, the db is already loaded as 4, so we won't do anything in load and just reuse.
##########
sql/core/src/test/scala/org/apache/spark/sql/execution/streaming/state/StatePartitionAllColumnFamiliesWriterSuite.scala:
##########
@@ -95,11 +101,29 @@ class StatePartitionAllColumnFamiliesWriterSuite extends StateDataSourceTestBase
transformFunc = None,
writeCheckpointMetadata = Some(targetCheckpointMetadata)
)
- rewriter.run()
+ val checkpointInfos = rewriter.run()
+ assert(spark.conf.get(SQLConf.STATE_STORE_CHECKPOINT_FORMAT_VERSION.key) == previousCkptVersion)
- // Commit to commitLog
+ // Commit to commitLog with checkpoint IDs
val latestCommit = targetCheckpointMetadata.commitLog.get(lastBatch).get
- targetCheckpointMetadata.commitLog.add(writeBatchId, latestCommit)
+ val operatorId = 0L
Review Comment:
why only for operator 0?
##########
sql/core/src/test/scala/org/apache/spark/sql/execution/streaming/state/RocksDBSuite.scala:
##########
@@ -3943,47 +3943,55 @@ class RocksDBSuite extends AlsoTestWithRocksDBFeatures with SharedSparkSession
}}
}
- test("SPARK-54420: load with createEmpty creates empty store") {
- val remoteDir = Utils.createTempDir().toString
- new File(remoteDir).delete()
-
- withDB(remoteDir) { db =>
- // loading batch 0 with loadEmpty = true
- db.load(0, None, loadEmpty = true)
- assert(iterator(db).isEmpty)
- db.put("a", "1")
- val (version1, _) = db.commit()
- assert(toStr(db.get("a")) === "1")
+ testWithStateStoreCheckpointIds(
+ "SPARK-54420: load with createEmpty creates empty store") { enableCkptId =>
+ val remoteDir = Utils.createTempDir().toString
+ new File(remoteDir).delete()
- // check we can load store normally even the previous one loadEmpty = true
- db.load(version1)
- db.put("b", "2")
- val (version2, _) = db.commit()
- assert(version2 === version1 + 1)
- assert(toStr(db.get("b")) === "2")
- assert(toStr(db.get("a")) === "1")
+ withDB(remoteDir, enableStateStoreCheckpointIds = enableCkptId) { db =>
+ // loading batch 0 with loadEmpty = true
+ db.load(0, None, loadEmpty = true)
+ assert(iterator(db).isEmpty)
+ db.put("a", "1")
+ val (version1, checkpointInfoV1) = db.commit()
+ assert(toStr(db.get("a")) === "1")
- // load an empty store
- db.load(version2, loadEmpty = true)
- db.put("c", "3")
- val (version3, _) = db.commit()
- assert(db.get("b") === null)
- assert(db.get("a") === null)
- assert(toStr(db.get("c")) === "3")
- assert(version3 === version2 + 1)
-
- // load 2 empty store in a row
- db.load(version3, loadEmpty = true)
- db.put("d", "4")
- val (version4, _) = db.commit()
- assert(db.get("c") === null)
- assert(toStr(db.get("d")) === "4")
- assert(version4 === version3 + 1)
-
- db.load(version4)
- db.put("e", "5")
- db.commit()
- assert(db.iterator().map(toStr).toSet === Set(("d", "4"), ("e", "5")))
+ // check we can load store normally even the previous one loadEmpty = true
+ db.load(version1, checkpointInfoV1.stateStoreCkptId)
+ db.put("b", "2")
+ val (version2, _) = db.commit()
+ assert(version2 === version1 + 1)
+ assert(toStr(db.get("b")) === "2")
+ assert(toStr(db.get("a")) === "1")
+
+ // load an empty store
+ db.load(version2, loadEmpty = true)
+ db.put("c", "3")
+ val (version3, _) = db.commit()
+ assert(db.get("b") === null)
+ assert(db.get("a") === null)
+ assert(toStr(db.get("c")) === "3")
+ assert(version3 === version2 + 1)
+
+ // load 2 empty store in a row
+ db.load(version3, loadEmpty = true)
+ db.put("d", "4")
+ val (version4, checkpointInfoV4) = db.commit()
+ assert(db.get("c") === null)
+ assert(toStr(db.get("d")) === "4")
+ assert(version4 === version3 + 1)
+
+ db.load(version4, checkpointInfoV4.stateStoreCkptId)
+ db.put("e", "5")
+ db.commit()
+ assert(db.iterator().map(toStr).toSet === Set(("d", "4"), ("e", "5")))
+
+ if (enableCkptId) {
+ val ex = intercept[IllegalArgumentException] {
+ db.load(version4, checkpointInfoV4.stateStoreCkptId, loadEmpty = true)
+ }
+ assert(ex.getMessage.contains("stateStoeCkptId should be empty when loadEmpty is true"))
Review Comment:
stateStoreCkptId
##########
sql/core/src/test/scala/org/apache/spark/sql/execution/streaming/state/OfflineStateRepartitionSuite.scala:
##########
@@ -454,4 +491,38 @@ object OfflineStateRepartitionTestUtils {
}
}
}
+
+ private def verifyCheckpointIds(
+ repartitionBatchId: Long,
+ checkpointMetadata: StreamingQueryCheckpointMetadata,
+ expectedShufflePartitions: Int,
+ expectedStoreCount: Int): Unit = {
+ // Verify commit log has the repartition batch with checkpoint IDs
+ val commitOpt = checkpointMetadata.commitLog.get(repartitionBatchId)
+ assert(commitOpt.isDefined, s"Commit for batch $repartitionBatchId should exist")
+
+ val commitMetadata = commitOpt.get
+
+ // Verify stateUniqueIds is present for checkpoint V2
+ assert(commitMetadata.stateUniqueIds.isDefined,
+ "stateUniqueIds should be present in commit metadata when checkpoint version >= 2")
+
+ val operatorIdToCkptInfos = commitMetadata.stateUniqueIds.get
+ assert(operatorIdToCkptInfos.nonEmpty,
+ "operatorIdToCkptInfos should not be empty")
+
+ // Verify structure for each operator
+ operatorIdToCkptInfos.foreach { case (operatorId, partitionToCkptIds) =>
+ // Should have checkpoint IDs for all partitions
+ assert(partitionToCkptIds.length == expectedShufflePartitions,
+ s"Operator $operatorId: Expected $expectedShufflePartitions partition checkpoint IDs, " +
+ s"but found ${partitionToCkptIds.length}")
+ // Each partition should have checkpoint IDs (at least one per store)
+ partitionToCkptIds.zipWithIndex.foreach { case (ckptIds, partitionId) =>
+ assert(ckptIds.length == expectedStoreCount,
Review Comment:
Each operator can have different store counts. So we need expectedStoreCount per operator
##########
sql/core/src/test/scala/org/apache/spark/sql/execution/streaming/state/StatePartitionAllColumnFamiliesWriterSuite.scala:
##########
@@ -95,11 +101,29 @@ class StatePartitionAllColumnFamiliesWriterSuite extends StateDataSourceTestBase
transformFunc = None,
writeCheckpointMetadata = Some(targetCheckpointMetadata)
)
- rewriter.run()
+ val checkpointInfos = rewriter.run()
+ assert(spark.conf.get(SQLConf.STATE_STORE_CHECKPOINT_FORMAT_VERSION.key) == previousCkptVersion)
- // Commit to commitLog
+ // Commit to commitLog with checkpoint IDs
val latestCommit = targetCheckpointMetadata.commitLog.get(lastBatch).get
- targetCheckpointMetadata.commitLog.add(writeBatchId, latestCommit)
+ val operatorId = 0L
+ val partitionSeq: Array[Array[StateStoreCheckpointInfo]] = checkpointInfos(operatorId)
+ val commitMetadata = if (StateStoreConf(conf).enableStateStoreCheckpointIds) {
+ // For join operators: 4 stores (left-keyToNumValues, left-keyWithIndexToValue,
+ // right-keyToNumValues, right-keyWithIndexToValue)
+ // For regular operators: 1 store
+ val ckptIds: Array[Array[String]] = partitionSeq.map { storesSeq =>
Review Comment:
ditto
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-54590][SS] Support CheckpointV2 for StatePartitionAllColumnFamiliesWriter [spark]
Posted by "zifeif2 (via GitHub)" <gi...@apache.org>.
zifeif2 commented on code in PR #53720:
URL: https://github.com/apache/spark/pull/53720#discussion_r2684474214
##########
sql/core/src/test/scala/org/apache/spark/sql/execution/streaming/state/OfflineStateRepartitionSuite.scala:
##########
@@ -312,8 +323,21 @@ object OfflineStateRepartitionTestUtils {
verifyOffsetAndCommitLog(
batchId, previousBatchId, expectedShufflePartitions, checkpointMetadata)
verifyPartitionDirs(checkpointLocation, expectedShufflePartitions)
+
+ val serializableConf = new SerializableConfiguration(hadoopConf)
+ val baseOperatorsMetadata = getOperatorMetadata(
+ checkpointLocation, serializableConf, previousBatchId)
+ val repartitionOperatorsMetadata = getOperatorMetadata(
Review Comment:
move the construction of operatorsMetadata out of verifyOperatorMetadata because we need to get the number of state store to verifyCheckpointIds
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/state/StateRewriter.scala:
##########
@@ -81,24 +82,46 @@ class StateRewriter(
private val stateRootLocation = new Path(
resolvedCheckpointLocation, StreamingCheckpointConstants.DIR_NAME_STATE).toString
- def run(): Unit = {
+ // return a Map[operator id, Array[partition -> Array[stateStore -> StateStoreCheckpointInfo]]]
+ def run(): Map[Long, Array[Array[StateStoreCheckpointInfo]]] = {
logInfo(log"Starting state rewrite for " +
log"checkpointLocation=${MDC(CHECKPOINT_LOCATION, resolvedCheckpointLocation)}, " +
log"readCheckpointLocation=" +
log"${MDC(CHECKPOINT_LOCATION, readResolvedCheckpointLocation.getOrElse(""))}, " +
log"readBatchId=${MDC(BATCH_ID, readBatchId)}, " +
log"writeBatchId=${MDC(BATCH_ID, writeBatchId)}")
- val (_, timeTakenMs) = Utils.timeTakenMs {
+ val (checkpointInfos, timeTakenMs) = Utils.timeTakenMs {
runInternal()
}
+ val allCheckpointInfos = checkpointInfos.values.flatMap { checkpointInfos =>
+ checkpointInfos.flatten.toSeq
Review Comment:
I tried printing out checkpointInfos as is, but it print out something like:
```
Map(0 -> [[Lorg.apache.spark.sql.execution.streaming.state.StateStoreCheckpointInfo;@28d6bdea))
```
while if we flatten it and print, it will can printout the detailed info
```
List(StateStoreCheckpointInfo(0,3,Some(3b36fc9a-128e-4aae-a9ea-dc743ba87c98),None), StateStoreCheckpointInfo(1,3,Some(2537e690-860a-4662-a526-3ea7346a01b4),None))
```
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/state/OfflineStateRepartitionRunner.scala:
##########
@@ -289,12 +290,27 @@ class OfflineStateRepartitionRunner(
}
}
- private def commitBatch(newBatchId: Long, lastCommittedBatchId: Long): Unit = {
+ private def commitBatch(
+ newBatchId: Long,
+ lastCommittedBatchId: Long,
+ opIdToStateStoreCkptInfo: Map[Long, Array[Array[StateStoreCheckpointInfo]]]): Unit = {
+ val enableCheckpointId = StatefulOperatorStateInfo.
+ enableStateStoreCheckpointIds(sparkSession.sessionState.conf)
val latestCommit = checkpointMetadata.commitLog.get(lastCommittedBatchId).get
+ val commitMetadata = if (enableCheckpointId) {
+ val opIdToPartitionCkptInfo: Map[Long, Array[Array[String]]] =
+ opIdToStateStoreCkptInfo.map {
+ case (operator, partitionSeq) =>
+ operator -> partitionSeq.map(storeSeq =>
+ storeSeq.flatMap(info => info.stateStoreCkptId))
Review Comment:
I see. I was thinking we might need other fields in StateCheckpointInfo from StateRewriter in future so I kept StateRewriter returning the entire StateStoreCkptInfo instead of just the checkpoint ids. Down to just let StateRewriter return StateStoreCkptId though!
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-54590][SS] Support Checkpoint V2 for State Rewriter and Repartitioning [spark]
Posted by "zifeif2 (via GitHub)" <gi...@apache.org>.
zifeif2 commented on code in PR #53720:
URL: https://github.com/apache/spark/pull/53720#discussion_r2695731278
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/state/StateRewriter.scala:
##########
@@ -342,6 +378,37 @@ class StateRewriter(
None
}
}
+
+ private def verifyCheckpointVersion(): Unit = {
+ // Verify checkpoint version in sqlConf based on commitLog for readCheckpoint
+ // in case user forgot to set STATE_STORE_CHECKPOINT_FORMAT_VERSION.
+ // Using read batch commit since the latest commit could be a skipped batch.
+ // If SQLConf.STATE_STORE_CHECKPOINT_FORMAT_VERSION is wrong, readCheckpoint.commitLog
+ // will throw an exception, and we will propagate this exception upstream.
+ // This prevents the StateRewriter from failing to write the correct state files
+ try {
+ val writeCheckpoint =
Review Comment:
Chatted with Micheal offline, we are okay with using lazy readCheckpoint for now. The goal for now is to verify `CHECKPOINT_FORMAT_VERSION ` is correct before writing state
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-54590][SS] Support Checkpoint V2 for State Rewriter and Repartitioning [spark]
Posted by "zifeif2 (via GitHub)" <gi...@apache.org>.
zifeif2 commented on code in PR #53720:
URL: https://github.com/apache/spark/pull/53720#discussion_r2696105747
##########
common/utils/src/main/resources/error/error-conditions.json:
##########
@@ -5603,6 +5603,13 @@
"Ensure that the checkpoint is for a stateful streaming query and the query ran on a Spark version that supports operator metadata (Spark 4.0+)."
]
},
+ "STATE_CHECKPOINT_FORMAT_VERSION_MISMATCH" : {
+ "message" : [
+ "The checkpoint format version in SQLConf does not match the checkpoint version in the commit log.",
+ "Expected version v<expectedVersion>, but found v<actualVersion>.",
Review Comment:
Hmm could you specify where the typo is? Thanks!
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org