micheal feedback

sql/catalyst/src/main/scala/org/apache/spark/sql/internal/SQLConf.scala

@@ -2211,7 +2211,7 @@ object SQLConf {
       .version("4.0.0")
       .stringConf
       .checkValue(v => Set("UnsafeRow", "Avro").contains(v),
-        "Valid versions are 'UnsafeRow' and 'Avro'")
+        "Valid values are 'UnsafeRow' and 'Avro'")
       .createWithDefault("UnsafeRow")
 
   // The feature is still in development, so it is still internal.

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

@@ -32,6 +32,7 @@ import org.apache.spark.sql.types.StructType
  * @param stateName - name of logical state partition
  * @param keyExprEnc - Spark SQL encoder for key
  * @param valEncoder - Spark SQL encoder for value
+ * @param avroEnc: Optional Avro encoder and decoder to convert between S and Avro row
  * @param metrics - metrics to be updated as part of stateful processing
  * @tparam S - data type of object that will be stored in the list
  */

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

@@ -36,6 +36,7 @@ import org.apache.spark.util.NextIterator
  * @param valEncoder - Spark SQL encoder for value
  * @param ttlConfig  - TTL configuration for values  stored in this state
  * @param batchTimestampMs - current batch processing timestamp.
+ * @param avroEnc: Optional Avro encoder and decoder to convert between S and Avro row
  * @param metrics - metrics to be updated as part of stateful processing
  * @tparam S - data type of object that will be stored
  */

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

@@ -32,6 +32,7 @@ import org.apache.spark.sql.types.StructType
  * @param stateName - name of logical state partition
  * @param keyExprEnc - Spark SQL encoder for key
  * @param valEncoder - Spark SQL encoder for value
+ * @param avroEnc: Optional Avro encoder and decoder to convert between S and Avro row
  * @param metrics - metrics to be updated as part of stateful processing
  * @tparam K - type of key for map state variable
  * @tparam V - type of value for map state variable

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

@@ -36,6 +36,7 @@ import org.apache.spark.util.NextIterator
  * @param valEncoder - SQL encoder for state variable
  * @param ttlConfig  - the ttl configuration (time to live duration etc.)
  * @param batchTimestampMs - current batch processing timestamp.
+ * @param avroEnc: Optional Avro encoder and decoder to convert between S and Avro row
  * @param metrics - metrics to be updated as part of stateful processing
  * @tparam K - type of key for map state variable
  * @tparam V - type of value for map state variable

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

@@ -65,6 +65,14 @@ object TransformWithStateKeyValueRowSchemaUtils {
   }
 }
 
+/**
+ * A trait that defines encoding and decoding operations for state types in Structured Streaming.
+ * This encoder handles the conversion between user-defined types and storage types for both
+ * state keys and values, with optional TTL (Time-To-Live) support.
+ *
+ * @tparam V The user-defined value type to be stored in state
+ * @tparam S The storage type used to represent the state (e.g., UnsafeRow or Array[Byte])
+ */
 trait StateTypesEncoder[V, S] {
   def encodeGroupingKey(): S
   def encodeValue(value: V): S
@@ -78,7 +86,7 @@ trait StateTypesEncoder[V, S] {
  * Helper class providing APIs to encode the grouping key, and user provided values
  * to Spark [[UnsafeRow]].
  *
- * CAUTION: StateTypesEncoder class instance is *not* thread-safe.
+ * CAUTION: UnsafeRowTypesEncoder class instance is *not* thread-safe.
  * This class reuses the keyProjection and valueProjection for encoding grouping
  * key and state value respectively. As UnsafeProjection is not thread safe, this
  * class is also not thread safe.
@@ -171,16 +179,13 @@ object UnsafeRowTypesEncoder {
 
 /**
  * Helper class providing APIs to encode the grouping key, and user provided values
- * to Spark [[UnsafeRow]].
- *
- * CAUTION: StateTypesEncoder class instance is *not* thread-safe.
- * This class reuses the keyProjection and valueProjection for encoding grouping
- * key and state value respectively. As UnsafeProjection is not thread safe, this
- * class is also not thread safe.
+ * to an Avro Byte Array.
  *
  * @param keyEncoder - SQL encoder for the grouping key, key type is implicit
  * @param valEncoder - SQL encoder for value of type `S`
  * @param stateName - name of logical state partition
+ * @param hasTtl - whether or not TTL is enabled for this state variable
+ * @param avroEnc = Avro encoder that should be specified to encode keys and values to byte arrays
  * @tparam V - value type
  */
 class AvroTypesEncoder[V](

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

@@ -732,7 +732,8 @@ object TransformWithStateExec {
       queryRunId = UUID.randomUUID(),
       operatorId = 0,
       storeVersion = 0,
-      numPartitions = shufflePartitions
+      numPartitions = shufflePartitions,
+      stateStoreCkptIds = None
     )
 
     new TransformWithStateExec(

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

@@ -32,6 +32,7 @@ import org.apache.spark.sql.streaming.ValueState
  * @param keyExprEnc - Spark SQL encoder for key
  * @param valEncoder - Spark SQL encoder for value
  * @param metrics - metrics to be updated as part of stateful processing
+ * @param avroEnc: Optional Avro encoder and decoder to convert between S and Avro row
  * @tparam S - data type of object that will be stored
  */
 class ValueStateImpl[S](

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

@@ -34,6 +34,7 @@ import org.apache.spark.sql.streaming.{TTLConfig, ValueState}
  * @param valEncoder - Spark SQL encoder for value
  * @param ttlConfig  - TTL configuration for values  stored in this state
  * @param batchTimestampMs - current batch processing timestamp.
+ * @param avroEnc: Optional Avro encoder and decoder to convert between S and Avro row
  * @param metrics - metrics to be updated as part of stateful processing
  * @tparam S - data type of object that will be stored
  */

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

@@ -102,7 +102,6 @@ private[sql] class HDFSBackedStateStoreProvider extends StateStoreProvider with
       throw StateStoreErrors.unsupportedOperationException("multipleValuesPerKey", "HDFSStateStore")
     }
 
-
     override def get(key: Array[Byte], colFamilyName: String): Array[Byte] = {
       throw StateStoreErrors.unsupportedOperationException("Byte array method", "HDFSStateStore")
     }