[SPARK-47570][SS] Integrate range scan encoder changes with timer implementation #45709

jingz-db · 2024-03-25T21:43:06Z

What changes were proposed in this pull request?

Previously timer state implementation was using No prefix rocksdb state encoder. When doing iterator() or prefix(), the returned iterator is not sorted on timestamp value. After Anish's PR for supporting range scan encoder, we could integrate it with TimerStateImpl such that we will use range scan encoder on timer to key.

Why are the changes needed?

The changes are part of the work around adding new stateful streaming operator for arbitrary state mgmt that provides a bunch of new features listed in the SPIP JIRA here - https://issues.apache.org/jira/browse/SPARK-45939

Does this PR introduce any user-facing change?

No

How was this patch tested?

Added unit tests in TimerSuite

Was this patch authored or co-authored using generative AI tooling?

No

HyukjinKwon · 2024-03-26T01:53:06Z

@jingz-db mind liking the JIRA into the PR title? See also https://spark.apache.org/contributing.html

jingz-db · 2024-03-26T17:48:33Z

@jingz-db mind liking the JIRA into the PR title? See also https://spark.apache.org/contributing.html

Just did. Thanks!

anishshri-db · 2024-03-26T18:52:27Z

sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/TransformWithStateExec.scala

@@ -161,25 +161,23 @@ case class TransformWithStateExec(
        assert(batchTimestampMs.isDefined)
        val batchTimestamp = batchTimestampMs.get
        val procTimeIter = processorHandle.getExpiredTimers()


Could we pass the timeout to the iterator and just return the expired rows instead ? i.e. move the filtering logic within the TimerStateImpl ?

sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/TransformWithStateExec.scala

anishshri-db · 2024-03-26T21:44:24Z

sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/TimerStateImpl.scala

-          val rowPair = iter.next()
-          val keyRow = rowPair.key
-          val result = getTimerRowFromSecIndex(keyRow)
+        val rowPair = if (iter.hasNext) iter.next() else null


Lets keep the original if condition as it is ? and add a condition to return null or result within this case ?

anishshri-db · 2024-03-26T21:46:33Z

...re/src/main/scala/org/apache/spark/sql/execution/streaming/StatefulProcessorHandleImpl.scala

@@ -164,11 +164,13 @@ class StatefulProcessorHandleImpl(

  /**
   * Function to retrieve all registered timers for all grouping keys
+   * @param expiryTimestampMs Threshold for expired timestamp in milliseconds, this function
+   *                          will return every timers that has (strictly) smaller timestamp


Nit: will return all timers that have timestamp less than passed threshold

anishshri-db · 2024-03-26T22:11:47Z

sql/core/src/test/scala/org/apache/spark/sql/execution/streaming/state/TimerSuite.scala

+      timerTimerStamps3.foreach(timerState3.registerTimer)
+      ImplicitGroupingKeyTracker.removeImplicitKey()
+
+      ImplicitGroupingKeyTracker.setImplicitKey("test_key1")


If we remove setting the implicit key just for this section (for the calls to getExpiredTimers) would the test still work ?

Yes it works. Will removed them in the next commit.

anishshri-db

lgtm

anishshri-db · 2024-03-26T22:21:05Z

sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/TimerStateImpl.scala

@@ -188,9 +187,12 @@ class TimerStateImpl(

  /**
   * Function to get all the registered timers for all grouping keys
+   * @param expiryTimestampMs Threshold for expired timestamp in milliseconds, this function


Nit: maybe add a small comment here mentioning that we perform a range scan and stop iterating once the key row timestamp exceeds the threshold

This class is not user-facing, right? If it is, I'd suggest avoiding implementation detail. Looks like as it doesn't seem to be an user facing, but just to remind.

Thanks for pointing this out (I did not realize this before but looks like we did the right thing)!

HeartSaVioR

+1

HeartSaVioR · 2024-03-27T04:37:32Z

Thanks! Merging to master.

…lementation ### What changes were proposed in this pull request? Previously timer state implementation was using No prefix rocksdb state encoder. When doing `iterator()` or `prefix()`, the returned iterator is not sorted on timestamp value. After Anish's PR for supporting range scan encoder, we could integrate it with `TimerStateImpl` such that we will use range scan encoder on `timer to key`. ### Why are the changes needed? The changes are part of the work around adding new stateful streaming operator for arbitrary state mgmt that provides a bunch of new features listed in the SPIP JIRA here - https://issues.apache.org/jira/browse/SPARK-45939 ### Does this PR introduce _any_ user-facing change? No ### How was this patch tested? Added unit tests in `TimerSuite` ### Was this patch authored or co-authored using generative AI tooling? No Closes apache#45709 from jingz-db/integrate-range-scan. Authored-by: jingz-db <jing.zhan@databricks.com> Signed-off-by: Jungtaek Lim <kabhwan.opensource@gmail.com>

github-actions bot added SQL STRUCTURED STREAMING DOCS labels Mar 25, 2024

jingz-db marked this pull request as ready for review March 25, 2024 22:24

rebase on master

1f128d5

jingz-db force-pushed the integrate-range-scan branch from 164928f to 1f128d5 Compare March 26, 2024 17:45

github-actions bot removed the DOCS label Mar 26, 2024

jingz-db changed the title ~~Integrate range scan encoder changes with timer implementation~~ [SPARK-47570][SS] Integrate range scan encoder changes with timer implementation Mar 26, 2024