ericm-db commented on code in PR #56332:
URL: https://github.com/apache/spark/pull/56332#discussion_r3616794911
##########
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:
Is there any way to consolidate the if-else statements here to one if else?
##########
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:
could we just pass in readOnly here instead of adding the forceSnapshot
boolean?
--
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]
