anishshri-db commented on code in PR #53911:
URL: https://github.com/apache/spark/pull/53911#discussion_r2725935234
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/state/RocksDBStateEncoder.scala:
##########
@@ -48,6 +48,9 @@ sealed trait RocksDBKeyStateEncoder {
def encodePrefixKey(prefixKey: UnsafeRow): Array[Byte]
def encodeKey(row: UnsafeRow): Array[Byte]
def decodeKey(keyBytes: Array[Byte]): UnsafeRow
+ def supportEventTime: Boolean
Review Comment:
Same here with all these methods
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/state/HDFSBackedStateStoreProvider.scala:
##########
@@ -323,6 +323,12 @@ private[sql] class HDFSBackedStateStoreProvider extends
StateStoreProvider with
key: UnsafeRow, values: Array[UnsafeRow], colFamilyName: String): Unit
= {
throw StateStoreErrors.unsupportedOperationException("mergeList",
providerName)
}
+
+ override def initiateEventTimeAwareStateOperations(
Review Comment:
The phrasing of this just seems a bit odd/non-conformant in this layer.
Could we rename/rephrase this ?
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/state/RocksDBStateEncoder.scala:
##########
@@ -328,6 +331,39 @@ trait DataEncoder {
*/
def encodePrefixKeyForRangeScan(row: UnsafeRow): Array[Byte]
+ /**
+ * Encodes key and event time, ensuring prefix scan with key and also proper
sort order with
+ * event time within the same key in RocksDB.
+ *
+ * This method handles the encoding as follows:
Review Comment:
Why not just expose this in a more generic fashion ?
--
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]
