liviazhu commented on code in PR #56332:
URL: https://github.com/apache/spark/pull/56332#discussion_r3625983282
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/operators/stateful/join/SymmetricHashJoinStateManager.scala:
##########
@@ -353,19 +400,34 @@ class SymmetricHashJoinStateManagerV4(
stateStoreProvider.getClass.toString)
}
val opts = handlerSnapshotOptions.get
- stateStoreProvider.asInstanceOf[SupportsFineGrainedReplay]
- .replayStateFromSnapshot(
+ val replayer =
stateStoreProvider.asInstanceOf[SupportsFineGrainedReplay]
+ if (readOnly) {
+ replayer.replayReadStateFromSnapshot(
opts.snapshotVersion,
opts.endVersion,
- readOnly = true,
opts.startStateStoreCkptId,
opts.endStateStoreCkptId)
+ } else {
+ replayer.replayStateFromSnapshot(
+ opts.snapshotVersion,
+ opts.endVersion,
+ readOnly = false,
+ opts.startStateStoreCkptId,
+ opts.endStateStoreCkptId)
+ }
+ } else if (readOnly) {
+ stateStoreProvider.getReadStore(stateInfo.get.storeVersion,
stateStoreCkptId)
} else {
stateStoreProvider.getStore(stateInfo.get.storeVersion,
stateStoreCkptId)
}
}
- logInfo(log"Loaded store ${MDC(STATE_STORE_ID, store.id)}")
- store
+ if (readOnly) {
+ _readStateStore = handle
+ logInfo(log"Loaded read-only store ${MDC(STATE_STORE_ID, handle.id)}")
+ } else {
+ _stateStore = handle.asInstanceOf[StateStore]
+ logInfo(log"Loaded store ${MDC(STATE_STORE_ID, handle.id)}")
+ }
Review Comment:
Hm can't think of a way to do this that makes it less complicated.
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/state/RocksDBStateStoreProvider.scala:
##########
@@ -201,16 +201,44 @@ private[sql] class RocksDBStateStoreProvider
}
}
+ // Read-only registration: sets up the virtual column-family id and the
key/value encoders
+ // needed to decode this family. Purely in-memory -- it does not write to
the checkpoint and
+ // carries no write intent (no forced snapshot). Safe to call on a
read-only store.
+ override def registerColFamily(
+ colFamilyName: String,
+ keySchema: StructType,
+ valueSchema: StructType,
+ keyStateEncoderSpec: KeyStateEncoderSpec,
+ useMultipleValuesPerKey: Boolean = false,
+ isInternal: Boolean = false): Unit = {
+ registerColFamilyInternal(colFamilyName, keySchema, valueSchema,
keyStateEncoderSpec,
+ useMultipleValuesPerKey, isInternal, forceSnapshot = false)
+ }
+
override def createColFamilyIfAbsent(
colFamilyName: String,
keySchema: StructType,
valueSchema: StructType,
keyStateEncoderSpec: KeyStateEncoderSpec,
useMultipleValuesPerKey: Boolean = false,
isInternal: Boolean = false): Unit = {
+ // Writable variant: same registration plus write intent -- a newly
created family forces a
+ // snapshot on commit so it is durably persisted.
+ registerColFamilyInternal(colFamilyName, keySchema, valueSchema,
keyStateEncoderSpec,
+ useMultipleValuesPerKey, isInternal, forceSnapshot = true)
+ }
+
+ private def registerColFamilyInternal(
+ colFamilyName: String,
+ keySchema: StructType,
+ valueSchema: StructType,
+ keyStateEncoderSpec: KeyStateEncoderSpec,
+ useMultipleValuesPerKey: Boolean,
+ isInternal: Boolean,
+ forceSnapshot: Boolean): Unit = {
validateAndTransitionState(UPDATE)
verifyColFamilyCreationOrDeletion("create_col_family", colFamilyName,
isInternal)
- val cfId = rocksDB.createColFamilyIfAbsent(colFamilyName, isInternal)
+ val cfId = rocksDB.createColFamilyIfAbsent(colFamilyName, isInternal,
forceSnapshot)
Review Comment:
Done
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
