anishshri-db commented on code in PR #47445:
URL: https://github.com/apache/spark/pull/47445#discussion_r1687021568
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/IncrementalExecution.scala:
##########
@@ -208,14 +208,25 @@ class IncrementalExecution(
}
val schemaValidationResult = statefulOp.
validateAndMaybeEvolveStateSchema(hadoopConf, currentBatchId,
stateSchemaVersion)
+ val stateSchemaPaths = schemaValidationResult.map(_.schemaPath)
// write out the state schema paths to the metadata file
statefulOp match {
case stateStoreWriter: StateStoreWriter =>
- val metadata = stateStoreWriter.operatorStateMetadata()
- // TODO: [SPARK-48849] Populate metadata with stateSchemaPaths if
metadata version is v2
- val metadataWriter = new OperatorStateMetadataWriter(new Path(
- checkpointLocation,
stateStoreWriter.getStateInfo.operatorId.toString), hadoopConf)
- metadataWriter.write(metadata)
+ val metadata =
stateStoreWriter.operatorStateMetadata(stateSchemaPaths)
+ stateStoreWriter match {
+ case tws: TransformWithStateExec =>
Review Comment:
Can we just do similar to pattern above for state schema ? Lets parse the
operator metadata version and pass it to `metadataWriter.write` and we can just
consolidate the write implementation in that file itself ?
--
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]
