[SPARK-48772][SS][SQL] State Data Source Change Feed Reader Mode

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

@@ -471,12 +471,17 @@ trait SupportsFineGrainedReplay {
   }
 
   /**
-   * Return a [[StateStoreChangeDataReader]] that reads the changelogs entries from startVersion to
+   * Return an iterator that reads all the entries of changelogs from startVersion to
    * endVersion.
+   * Each record is represented by a tuple of (recordType: [[RecordType.Value]], key: [[UnsafeRow]],
+   * value: [[UnsafeRow]], batchId: [[Long]])
+   * A put record is returned as a tuple(recordType, key, value, batchId)
+   * A delete record is return as a tuple(recordType, key, null, batchId)
    *
    * @param startVersion starting changelog version
    * @param endVersion ending changelog version
-   * @return
+   * @return tuple(recordType: [[RecordType.Value]], nested key: [[UnsafeRow]],
+   *         nested value: [[UnsafeRow]], batchId: [[Long]])
    */
   def getStateStoreChangeDataReader(startVersion: Long, endVersion: Long):
     NextIterator[(RecordType.Value, UnsafeRow, UnsafeRow, Long)]

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

@@ -403,7 +403,7 @@ class StateStoreChangelogReaderV2(
 
 /**
  * Base class representing a iterator that iterates over a range of changelog files in a state
- * store. In each iteration, it will return a tuple of (changeType: [[RecordType]],
+ * store. In each iteration, it will return a ByteArrayPair of (changeType: [[RecordType]],
  * nested key: [[UnsafeRow]], nested value: [[UnsafeRow]], batchId: [[Long]])
  *
  * @param fm checkpoint file manager used to manage streaming query checkpoint