removing ': Array[StateStoreMetadata]'

sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/state/metadata/StateMetadataSource.scala

@@ -212,20 +212,31 @@ class StateMetadataPartitionReader(
   private[state] lazy val stateMetadata: Iterator[StateMetadataTableEntry] = {
     allOperatorStateMetadata.flatMap { operatorStateMetadata =>
       require(operatorStateMetadata.version == 1 || operatorStateMetadata.version == 2)
-      val operatorProperties = operatorStateMetadata match {
-        case _: OperatorStateMetadataV1 => ""
-        case v2: OperatorStateMetadataV2 => v2.operatorPropertiesJson
-      }
-      operatorStateMetadata.stateStoreInfo.map { stateStoreMetadata =>
-        StateMetadataTableEntry(operatorStateMetadata.operatorInfo.operatorId,
-          operatorStateMetadata.operatorInfo.operatorName,
-          stateStoreMetadata.storeName,
-          stateStoreMetadata.numPartitions,
-          if (batchIds.nonEmpty) batchIds.head else -1,
-          if (batchIds.nonEmpty) batchIds.last else -1,
-          operatorProperties,
-          stateStoreMetadata.numColsPrefixKey
-        )
+      operatorStateMetadata match {
+        case v1: OperatorStateMetadataV1 =>
+            v1.stateStoreInfo.map { stateStoreMetadata =>
+                StateMetadataTableEntry(v1.operatorInfo.operatorId,
+                v1.operatorInfo.operatorName,
+                stateStoreMetadata.storeName,
+                stateStoreMetadata.numPartitions,
+                if (batchIds.nonEmpty) batchIds.head else -1,
+                if (batchIds.nonEmpty) batchIds.last else -1,
+                "",
+                stateStoreMetadata.numColsPrefixKey
+                )
+            }
+        case v2: OperatorStateMetadataV2 =>
+          v2.stateStoreInfo.map { stateStoreMetadata =>
+            StateMetadataTableEntry(v2.operatorInfo.operatorId,
+              v2.operatorInfo.operatorName,
+              stateStoreMetadata.storeName,
+              stateStoreMetadata.numPartitions,
+              if (batchIds.nonEmpty) batchIds.head else -1,
+              if (batchIds.nonEmpty) batchIds.last else -1,
+              v2.operatorPropertiesJson,
+              stateStoreMetadata.numColsPrefixKey
+            )
+          }
       }
     }
   }.iterator

sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/StreamingSymmetricHashJoinExec.scala

@@ -230,7 +230,7 @@ case class StreamingSymmetricHashJoinExec(
   override def operatorStateMetadata(): OperatorStateMetadata = {
     val info = getStateInfo
     val operatorInfo = OperatorInfoV1(info.operatorId, shortName)
-    val stateStoreInfo: Array[StateStoreMetadata] =
+    val stateStoreInfo =
       stateStoreNames.map(StateStoreMetadataV1(_, 0, info.numPartitions)).toArray
     OperatorStateMetadataV1(operatorInfo, stateStoreInfo)
   }

sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/TransformWithStateExec.scala

@@ -428,7 +428,7 @@ case class TransformWithStateExec(
     val operatorInfo = OperatorInfoV1(info.operatorId, shortName)
     // stateSchemaFilePath should be populated at this point
     assert(info.stateSchemaPath.isDefined)
-    val stateStoreInfo: Array[StateStoreMetadata] =
+    val stateStoreInfo =
       Array(StateStoreMetadataV2(
         StateStoreId.DEFAULT_STORE_NAME, 0, info.numPartitions, info.stateSchemaPath.get))
 

sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/state/OperatorStateMetadata.scala

@@ -71,19 +71,17 @@ trait OperatorStateMetadata {
   def version: Int
 
   def operatorInfo: OperatorInfo
-
-  def stateStoreInfo: Array[StateStoreMetadata]
 }
 
 case class OperatorStateMetadataV1(
     operatorInfo: OperatorInfoV1,
-    stateStoreInfo: Array[StateStoreMetadata]) extends OperatorStateMetadata {
+    stateStoreInfo: Array[StateStoreMetadataV1]) extends OperatorStateMetadata {
   override def version: Int = 1
 }
 
 case class OperatorStateMetadataV2(
     operatorInfo: OperatorInfoV1,
-    stateStoreInfo: Array[StateStoreMetadata],
+    stateStoreInfo: Array[StateStoreMetadataV2],
     operatorPropertiesJson: String) extends OperatorStateMetadata {
   override def version: Int = 2
 }

sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/statefulOperators.scala

@@ -201,7 +201,7 @@ trait StateStoreWriter extends StatefulOperator with PythonSQLMetrics { self: Sp
   def operatorStateMetadata(): OperatorStateMetadata = {
     val info = getStateInfo
     val operatorInfo = OperatorInfoV1(info.operatorId, shortName)
-    val stateStoreInfo: Array[StateStoreMetadata] =
+    val stateStoreInfo =
       Array(StateStoreMetadataV1(StateStoreId.DEFAULT_STORE_NAME, 0, info.numPartitions))
     OperatorStateMetadataV1(operatorInfo, stateStoreInfo)
   }
@@ -922,7 +922,7 @@ case class SessionWindowStateStoreSaveExec(
   override def operatorStateMetadata(): OperatorStateMetadata = {
     val info = getStateInfo
     val operatorInfo = OperatorInfoV1(info.operatorId, shortName)
-    val stateStoreInfo: Array[StateStoreMetadata] = Array(StateStoreMetadataV1(
+    val stateStoreInfo = Array(StateStoreMetadataV1(
       StateStoreId.DEFAULT_STORE_NAME, stateManager.getNumColsForPrefixKey, info.numPartitions))
     OperatorStateMetadataV1(operatorInfo, stateStoreInfo)
   }

sql/core/src/test/scala/org/apache/spark/sql/execution/streaming/state/OperatorStateMetadataSuite.scala

@@ -106,7 +106,7 @@ class OperatorStateMetadataSuite extends StreamTest with SharedSparkSession {
         StopStream
       )
 
-      val expectedStateStoreInfo: Array[StateStoreMetadata] = Array(
+      val expectedStateStoreInfo = Array(
         StateStoreMetadataV1("left-keyToNumValues", 0, numShufflePartitions),
         StateStoreMetadataV1("left-keyWithIndexToValue", 0, numShufflePartitions),
         StateStoreMetadataV1("right-keyToNumValues", 0, numShufflePartitions),