You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@spark.apache.org by ka...@apache.org on 2026/10/22 01:35:29 UTC
(spark) branch branch-4.x updated: [SPARK-60348][SS] Fix state data source change feed dropping list state values after the first
This is an automated email from the ASF dual-hosted git repository.
HeartSaVioR pushed a commit to branch branch-4.x
in repository https://gitbox.apache.org/repos/asf/spark.git
The following commit(s) were added to refs/heads/branch-4.x by this push:
new dda5239d3efc [SPARK-60348][SS] Fix state data source change feed dropping list state values after the first
dda5239d3efc is described below
commit dda5239d3efca4acebee53287d5380d25d0456fc
Author: Kavpreet Grewal <ka...@gmail.com>
AuthorDate: Thu Oct 22 10:34:44 2026 +0900
[SPARK-60348][SS] Fix state data source change feed dropping list state values after the first
### What changes were proposed in this pull request?
Since SPARK-54118, `ListState.put` and `appendList` write all of their values in a single changelog record, but `RocksDBStateStoreChangeDataReader` decoded only the first value of each record.
The reader now returns one row per value of a multi-valued record. The first value keeps the record's change type and the rest are returned as `append`, which matches the change feed from before SPARK-54118.
### Why are the changes needed?
Reading list state with `readChangeFeed` silently drops every value after the first one. For example, `put(Array("a", "b", "c"))` shows up only as `update a`.
### Does this PR introduce _any_ user-facing change?
Yes. Reading list state with `readChangeFeed` now returns a row for every value written by `put` or `appendList`, instead of only the first one. This restores the behavior from before Spark 4.1.0.
### How was this patch tested?
Added a test to `StateDataSourceTransformWithStateSuite` that writes list state with `put` and `appendList` and reads it with the change feed, under both RocksDB merge operator versions. It fails without this fix.
### Was this patch authored or co-authored using generative AI tooling?
Generated-by: Claude Code (Claude Opus 5.5)
This pull request and its description were written by Isaac.
Closes #59521 from kavpreetgrewal/tws-list-state-cdf-multi-values.
Authored-by: Kavpreet Grewal <ka...@gmail.com>
Signed-off-by: Jungtaek Lim <ka...@gmail.com>
(cherry picked from commit 4a18a2e73962018bb9fc5e9966fb7b5691b709e9)
Signed-off-by: Jungtaek Lim <ka...@gmail.com>
---
.../state/RocksDBStateStoreProvider.scala | 23 ++++++-
.../StateDataSourceTransformWithStateSuite.scala | 70 ++++++++++++++++++++++
2 files changed, 91 insertions(+), 2 deletions(-)
diff --git a/sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/state/RocksDBStateStoreProvider.scala b/sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/state/RocksDBStateStoreProvider.scala
index 79c69634509b..24e91cf995aa 100644
--- a/sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/state/RocksDBStateStoreProvider.scala
+++ b/sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/state/RocksDBStateStoreProvider.scala
@@ -1667,7 +1667,18 @@ class RocksDBStateStoreChangeDataReader(
override protected val changelogSuffix: String = "changelog"
+ // A list put/merge writes all of its values in a single record. These hold the values of that
+ // record which are not returned yet, so that each value is returned as its own change row.
+ private var pendingValues: Iterator[UnsafeRow] = Iterator.empty
+ private var pendingKeyRow: UnsafeRow = _
+ private var pendingBatchId: Long = _
+
override def getNext(): (RecordType.Value, UnsafeRow, UnsafeRow, Long) = {
+ if (pendingValues.hasNext) {
+ // Values after the first one are appended to the list.
+ return (RecordType.MERGE_RECORD, pendingKeyRow, pendingValues.next(), pendingBatchId)
+ }
+
var currRecord: (RecordType.Value, Array[Byte], Array[Byte]) = null
val currEncoder: (RocksDBKeyStateEncoder, RocksDBValueStateEncoder, Short) =
keyValueEncoderMap.get(colFamilyNameOpt
@@ -1741,11 +1752,19 @@ class RocksDBStateStoreChangeDataReader(
}
val keyRow = currEncoder._1.decodeKey(currRecord._2)
+ val batchId = currentChangelogVersion - 1
if (currRecord._3 == null) {
- (currRecord._1, keyRow, null, currentChangelogVersion - 1)
+ (currRecord._1, keyRow, null, batchId)
+ } else if (currEncoder._2.supportsMultipleValuesPerKey) {
+ val values = currEncoder._2.decodeValues(currRecord._3)
+ val firstValue = values.next()
+ pendingValues = values
+ pendingKeyRow = keyRow
+ pendingBatchId = batchId
+ (currRecord._1, keyRow, firstValue, batchId)
} else {
val valueRow = currEncoder._2.decodeValue(currRecord._3)
- (currRecord._1, keyRow, valueRow, currentChangelogVersion - 1)
+ (currRecord._1, keyRow, valueRow, batchId)
}
}
}
diff --git a/sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/v2/state/StateDataSourceTransformWithStateSuite.scala b/sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/v2/state/StateDataSourceTransformWithStateSuite.scala
index 78e394b7cf75..b04c08d022e4 100644
--- a/sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/v2/state/StateDataSourceTransformWithStateSuite.scala
+++ b/sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/v2/state/StateDataSourceTransformWithStateSuite.scala
@@ -102,6 +102,30 @@ class SessionGroupsStatefulProcessor extends
}
}
+class SessionGroupsPutAndAppendListProcessor extends
+ StatefulProcessor[String, (String, String), String] {
+ @transient private var _groupsList: ListState[String] = _
+
+ override def init(
+ outputMode: OutputMode,
+ timeMode: TimeMode): Unit = {
+ _groupsList = getHandle.getListState("groupsList", Encoders.STRING, TTLConfig.NONE)
+ }
+
+ override def handleInputRows(
+ key: String,
+ inputRows: Iterator[(String, String)],
+ timerValues: TimerValues): Iterator[String] = {
+ val groups = inputRows.map(_._2).toArray.sorted
+ if (_groupsList.exists()) {
+ _groupsList.appendList(groups)
+ } else {
+ _groupsList.put(groups)
+ }
+ Iterator.empty
+ }
+}
+
class SessionGroupsStatefulProcessorWithTTL extends
StatefulProcessor[String, (String, String), String] {
@transient private var _groupsListWithTTL: ListState[String] = _
@@ -475,6 +499,52 @@ class StateDataSourceTransformWithStateSuite extends StateDataSourceTestBase
}
}
+ Seq(1, 2).foreach { mergeOperatorVersion =>
+ testWithChangelogCheckpointingEnabled("state data source cdf integration - list state " +
+ s"put and appendList (mergeOperatorVersion = $mergeOperatorVersion)") {
+ withTempDir { tempDir =>
+ withSQLConf(SQLConf.STATE_STORE_ROCKSDB_MERGE_OPERATOR_VERSION.key ->
+ mergeOperatorVersion.toString) {
+ val inputData = MemoryStream[(String, String)]
+ val result = inputData.toDS()
+ .groupByKey(x => x._1)
+ .transformWithState(new SessionGroupsPutAndAppendListProcessor(),
+ TimeMode.None(),
+ OutputMode.Update())
+
+ testStream(result, OutputMode.Update())(
+ StartStream(checkpointLocation = tempDir.getAbsolutePath),
+ AddData(inputData, ("session1", "group1"), ("session1", "group2"),
+ ("session2", "group3")),
+ CheckNewAnswer(),
+ AddData(inputData, ("session1", "group3"), ("session1", "group4"),
+ ("session2", "group5"), ("session2", "group6")),
+ CheckNewAnswer(),
+ StopStream
+ )
+
+ val changeFeedDf = spark.read
+ .format("statestore")
+ .option(StateSourceOptions.PATH, tempDir.getAbsolutePath)
+ .option(StateSourceOptions.STATE_VAR_NAME, "groupsList")
+ .option(StateSourceOptions.READ_CHANGE_FEED, true)
+ .option(StateSourceOptions.CHANGE_START_BATCH_ID, 0)
+ .load()
+ .selectExpr("batch_id", "change_type", "key.value", "list_element.value")
+
+ checkAnswer(changeFeedDf,
+ Seq(Row(0L, "update", "session1", "group1"),
+ Row(0L, "append", "session1", "group2"),
+ Row(0L, "update", "session2", "group3"),
+ Row(1L, "append", "session1", "group3"),
+ Row(1L, "append", "session1", "group4"),
+ Row(1L, "append", "session2", "group5"),
+ Row(1L, "append", "session2", "group6")))
+ }
+ }
+ }
+ }
+
test("state data source integration - list state and TTL") {
withTempDir { tempDir =>
withSQLConf(SQLConf.STATE_STORE_PROVIDER_CLASS.key ->
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@spark.apache.org
For additional commands, e-mail: commits-help@spark.apache.org