anishshri-db commented on code in PR #52820:
URL: https://github.com/apache/spark/pull/52820#discussion_r2482782836
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/state/RocksDB.scala:
##########
@@ -991,6 +992,70 @@ class RocksDB(
changelogWriter.foreach(_.put(keyWithPrefix, value))
}
+ /**
+ * Convert the given list of value row bytes into a single byte array. The
returned array
+ * bytes supports additional values to be later merged to it.
+ */
+ private def getListValuesInArrayByte(values: List[Array[Byte]]): Array[Byte]
= {
+
Review Comment:
nit: extra newline ?
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/state/RocksDBStateStoreProvider.scala:
##########
@@ -305,6 +330,28 @@ private[sql] class RocksDBStateStoreProvider
rocksDB.put(kvEncoder._1.encodeKey(key),
kvEncoder._2.encodeValue(value), colFamilyName)
}
+ override def putList(
+ key: UnsafeRow,
+ values: Array[UnsafeRow],
+ colFamilyName: String): Unit = {
+ validateAndTransitionState(UPDATE)
+ verify(state == UPDATING, "Cannot put after already committed or
aborted")
+ verify(key != null, "Key cannot be null")
+ require(values != null, "Cannot put a null value")
+ values.foreach(v => require(v != null, "Cannot put a null value in the
array"))
+ verifyColFamilyOperations("put", colFamilyName)
+
+ val kvEncoder = keyValueEncoderMap.get(colFamilyName)
+ verify(
+ kvEncoder._2.supportsMultipleValuesPerKey,
+ "Multiple values put operation requires an encoder" +
Review Comment:
nit: `Multi-value put operations`
##########
sql/core/src/test/scala/org/apache/spark/sql/execution/streaming/state/RocksDBSuite.scala:
##########
@@ -2015,6 +2015,58 @@ class RocksDBSuite extends AlsoTestWithRocksDBFeatures
with SharedSparkSession
}
}
+ test("RocksDB: ensure putList / mergeList operation correctness") {
Review Comment:
Do we have enough coverage for list state tests ?
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/state/StateStore.scala:
##########
@@ -208,6 +208,16 @@ trait StateStore extends ReadStateStore {
value: UnsafeRow,
colFamilyName: String = StateStore.DEFAULT_COL_FAMILY_NAME): Unit
+ /**
+ * Put a new list of non-null value for a non-null key. Implementations must
be aware that the
Review Comment:
nit: `non-null values for a`
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
