anishshri-db commented on code in PR #54278:
URL: https://github.com/apache/spark/pull/54278#discussion_r2814899564
##########
sql/core/src/test/scala/org/apache/spark/sql/execution/streaming/state/RocksDBTimestampEncoderOperationsSuite.scala:
##########
@@ -281,6 +281,70 @@ class RocksDBTimestampEncoderOperationsSuite extends
SharedSparkSession
}
}
+ test(s"Event time as prefix: iterator with multiple values (encoding =
$encoding)") {
Review Comment:
Does Avro work here or will be added later ? if so, can we add a TODO ?
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/state/StateStore.scala:
##########
@@ -171,10 +171,31 @@ trait ReadStateStore {
prefixKey: UnsafeRow,
colFamilyName: String = StateStore.DEFAULT_COL_FAMILY_NAME):
StateStoreIterator[UnsafeRowPair]
+ /**
+ * Return an iterator containing all the (key, value) pairs which are
matched with
+ * the given prefix key.
+ *
+ * It is expected to throw exception if Spark calls this method without
proper key encoding spec.
+ * It is also expected to throw exception if Spark calls this method without
setting
+ * multipleValuesPerKey as true for the column family.
Review Comment:
Where do we verify the `multipleValuesPerKey` condition ?
##########
sql/core/src/test/scala/org/apache/spark/sql/execution/streaming/state/RocksDBTimestampEncoderOperationsSuite.scala:
##########
@@ -281,6 +281,70 @@ class RocksDBTimestampEncoderOperationsSuite extends
SharedSparkSession
}
}
+ test(s"Event time as prefix: iterator with multiple values (encoding =
$encoding)") {
+ tryWithProviderResource(
+ newStoreProviderWithTimestampEncoder(
+ encoderType = "prefix",
+ useMultipleValuesPerKey = true,
+ dataEncoding = encoding)
+ ) { provider =>
+ val store = provider.getStore(0)
+
+ try {
+ // Put multiple values at different event times
+ // Insert in non-sorted order to verify ordering by event time
+ val values2 = Array(valueToRow(200), valueToRow(201))
+ store.putList(keyAndTimestampToRow("key1", 1, 1000L), values2)
Review Comment:
Lets add some different values here also ? similar to the encoder tests ?
##########
sql/core/src/test/scala/org/apache/spark/sql/execution/streaming/state/RocksDBTimestampEncoderOperationsSuite.scala:
##########
@@ -380,16 +528,18 @@ class RocksDBTimestampEncoderOperationsSuite extends
SharedSparkSession
val iter = encoderType match {
// For prefix encoder, we use iterator
case "prefix" =>
- store.iterator()
+ if (useMultipleValuesPerKey) store.iteratorWithMultiValues()
+ else store.iterator()
// For postfix encoder, we use prefix scan with ("key1", 1) as the
prefix key
case "postfix" =>
- store.prefixScan(keyToRow("key1", 1))
+ if (useMultipleValuesPerKey)
store.prefixScanWithMultiValues(keyToRow("key1", 1))
Review Comment:
same here ?
##########
sql/core/src/test/scala/org/apache/spark/sql/execution/streaming/state/RocksDBTimestampEncoderOperationsSuite.scala:
##########
@@ -380,16 +528,18 @@ class RocksDBTimestampEncoderOperationsSuite extends
SharedSparkSession
val iter = encoderType match {
// For prefix encoder, we use iterator
case "prefix" =>
- store.iterator()
+ if (useMultipleValuesPerKey) store.iteratorWithMultiValues()
Review Comment:
Could we use braces ?
--
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]
