renaming to suffix key

Author: ericm-db

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

@@ -62,7 +62,7 @@ class StateStoreColumnFamilySchemaUtils(initializeAvroSerde: Boolean) {
   private def getAvroSerde(
       keySchema: StructType,
       valSchema: StructType,
-      userKeySchema: Option[StructType] = None
+      suffixKeySchema: Option[StructType] = None
   ): Option[AvroEncoderSpec] = {
     if (initializeAvroSerde) {
       val avroType = SchemaConverters.toAvroType(valSchema)
@@ -76,18 +76,18 @@ class StateStoreColumnFamilySchemaUtils(initializeAvroSerde: Boolean) {
       val valueDeserializer = new AvroDeserializer(avroType, valSchema,
         avroOptions.datetimeRebaseModeInRead, avroOptions.useStableIdForUnionType,
         avroOptions.stableIdPrefixForUnionType, avroOptions.recursiveFieldMaxDepth)
-      val (userKeySerializer, userKeyDeserializer) = if (userKeySchema.isDefined) {
-        val userKeyAvroType = SchemaConverters.toAvroType(userKeySchema.get)
-        val ukSer = new AvroSerializer(userKeySchema.get, userKeyAvroType, nullable = false)
-        val ukDe = new AvroDeserializer(userKeyAvroType, userKeySchema.get,
+      val (suffixKeySer, suffixKeyDe) = if (suffixKeySchema.isDefined) {
+        val userKeyAvroType = SchemaConverters.toAvroType(suffixKeySchema.get)
+        val skSer = new AvroSerializer(suffixKeySchema.get, userKeyAvroType, nullable = false)
+        val skDe = new AvroDeserializer(userKeyAvroType, suffixKeySchema.get,
           avroOptions.datetimeRebaseModeInRead, avroOptions.useStableIdForUnionType,
           avroOptions.stableIdPrefixForUnionType, avroOptions.recursiveFieldMaxDepth)
-        (Some(ukSer), Some(ukDe))
+        (Some(skSer), Some(skDe))
       } else {
         (None, None)
       }
       Some(AvroEncoderSpec(
-        keySer, keyDe, valueSerializer, valueDeserializer, userKeySerializer, userKeyDeserializer))
+        keySer, keyDe, valueSerializer, valueDeserializer, suffixKeySer, suffixKeyDe))
     } else {
       None
     }

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

@@ -287,7 +287,7 @@ class PrefixKeyScanStateEncoder(
         ),
         encodeUnsafeRow(
           remainingKeyProjection(row),
-          avroEnc.get.userKeySerializer.get,
+          avroEnc.get.suffixKeySerializer.get,
           remainingKeyAvroType,
           out
         )
@@ -335,7 +335,7 @@ class PrefixKeyScanStateEncoder(
         ),
         decodeToUnsafeRow(
           remainingKeyEncoded,
-          avroEnc.get.userKeyDeserializer.get,
+          avroEnc.get.suffixKeyDeserializer.get,
           remainingKeyAvroType,
           remainingKeyProj
         )
@@ -703,7 +703,7 @@ class RangeKeyScanStateEncoder(
       val remainingEncoded = if (avroEnc.isDefined) {
         encodeUnsafeRow(
           remainingKeyProjection(row),
-          avroEnc.get.userKeySerializer.get,
+          avroEnc.get.suffixKeySerializer.get,
           remainingKeyAvroType,
           out
         )
@@ -767,7 +767,7 @@ class RangeKeyScanStateEncoder(
 
       val remainingKeyDecoded = if (avroEnc.isDefined) {
         decodeToUnsafeRow(remainingKeyEncoded,
-          avroEnc.get.userKeyDeserializer.get,
+          avroEnc.get.suffixKeyDeserializer.get,
           remainingKeyAvroType, remainingKeyAvroProjection)
       } else {
         decodeToUnsafeRow(remainingKeyEncoded,

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

@@ -45,8 +45,8 @@ case class AvroEncoderSpec(
   keyDeserializer: AvroDeserializer,
   valueSerializer: AvroSerializer,
   valueDeserializer: AvroDeserializer,
-  userKeySerializer: Option[AvroSerializer] = None,
-  userKeyDeserializer: Option[AvroDeserializer] = None
+  suffixKeySerializer: Option[AvroSerializer] = None,
+  suffixKeyDeserializer: Option[AvroDeserializer] = None
 ) extends Serializable
 
 // Used to represent the schema of a column family in the state store