address comment

Author: zifeif2

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

@@ -29,6 +29,13 @@ import org.apache.spark.sql.types.{NullType, StructField, StructType}
 import org.apache.spark.unsafe.types.UTF8String
 import org.apache.spark.util.{NextIterator, SerializableConfiguration}
 
+/**
+ * Information used specifically by StatePartitionAllColumnFamiliesReader
+ * @param colFamilySchemas a set of ColFamilySchema for all column families in an operator.
+ *                         The reader relies on this field to read data for all column family
+ * @param stateVariableInfos a list of TransformWithStateVariableInfo for state variables
+ *                           in TWS operator. The reader relies on this to check variable type
+ */
 case class AllColumnFamiliesReaderInfo(
     colFamilySchemas: Set[StateStoreColFamilySchema] = Set.empty,
     stateVariableInfos: List[TransformWithStateVariableInfo] = List.empty)
@@ -338,7 +345,7 @@ class StatePartitionAllColumnFamiliesReader(
           case StateStore.DEFAULT_COL_FAMILY_NAME => // createAndInit has registered default
           case _ =>
             val isInternal =
-              StateStoreColumnFamilySchemaUtils.isInternalColumn(cfSchema.colFamilyName)
+              StateStoreColumnFamilySchemaUtils.isInternalColFamily(cfSchema.colFamilyName)
             val useMultipleValuesPerKey = isListType(cfSchema.colFamilyName)
             require(cfSchema.keyStateEncoderSpec.isDefined,
               s"keyStateEncoderSpec must be defined for column family ${cfSchema.colFamilyName}")

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

@@ -99,7 +99,7 @@ object StateStoreColumnFamilySchemaUtils {
   def getStateNameFromCountIndexCFName(colFamilyName: String): String =
     getStateName(COUNT_INDEX_PREFIX, colFamilyName)
 
-  def isInternalColumn(colFamilyName: String): Boolean = {
+  def isInternalColFamily(colFamilyName: String): Boolean = {
     colFamilyName.startsWith("$")
   }
 
@@ -111,7 +111,7 @@ object StateStoreColumnFamilySchemaUtils {
    * @return true if this is an internal column family and Utils.isTesting is true
    */
   def isTestingInternalColFamily(colFamilyName: String): Boolean = {
-    org.apache.spark.util.Utils.isTesting && isInternalColumn(colFamilyName)
+    org.apache.spark.util.Utils.isTesting && isInternalColFamily(colFamilyName)
   }
 
   def getValueStateSchema[T](

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

@@ -1320,6 +1320,7 @@ class RocksDBSuite extends AlsoTestWithRocksDBFeatures with SharedSparkSession
         // Create a test column family
         val testCfName = "test_cf"
         db.createColFamilyIfAbsent(testCfName, isInternal = false)
+        assert(db.allColumnFamilyNames == Set(StateStore.DEFAULT_COL_FAMILY_NAME, testCfName))
 
         // Write initial data
         db.load(0)