HeartSaVioR commented on code in PR #44961:
URL: https://github.com/apache/spark/pull/44961#discussion_r1487111235
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/state/RocksDBStateStoreProvider.scala:
##########
@@ -65,6 +65,43 @@ private[sql] class RocksDBStateStoreProvider
value
}
+ override def valuesIterator(key: UnsafeRow, colFamilyName: String):
Iterator[UnsafeRow] = {
+ verify(key != null, "Key cannot be null")
+ verify(encoder.supportsMultipleValuesPerKey, "valuesIterator requires a
encoder " +
+ "that supports multiple values for a single key.")
+ val valueIterator =
encoder.decodeValues(rocksDB.get(encoder.encodeKey(key), colFamilyName))
+
+ if (!isValidated && valueIterator.nonEmpty) {
Review Comment:
historical reasoning - unfortunately we had two difference state store
checker being proposed at the same time. This is to cover the case with broken
UnsafeRow. I guess it's to cover the case of schema check with retaining the
characteristic of schema-on-read (which is arguably replaced by state schema
checker), but it still capture the case of incompatible UnsafeRow being broken
in any way.
Maybe checking a single key-value pair is not sufficient in any way though.
--
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]
