[SPARK-48511][SS] Remove TimeMode None from TransformWithState. #46825

sahnib · 2024-05-31T19:34:14Z

What changes were proposed in this pull request?

This PR removes the TimeMode None from supported time modes in TransformWithState operator. A structured streaming query works in either Processing time mode, or event Time mode depending on whether eventTime has been specified, hence this change aligns TimeMode properly with Streaming query semantics.

Note that if eventTimeColumn is specified for output dataset in transformWithState operator. TimeMode defaults to EventTime.

Why are the changes needed?

These changes are needed to align TimeMode values with how time flows in Streaming query.

Does this PR introduce any user-facing change?

Yes, modifies the TimeMode semantic for transformWithState.

How was this patch tested?

All existing unit test pass.

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

No

HyukjinKwon · 2024-06-03T01:55:02Z

Mind filing a JIRA please?

HyukjinKwon · 2024-06-03T01:55:39Z

sql/api/src/main/java/org/apache/spark/sql/streaming/TimeMode.java

When did we add this? If this is already released out, we can't just remove and break it.

We added this on April 7, 2024 as part of TTL support for Streaming State with TransformWithState operator. This operator is a new addition to Spark Structured Streaming, and is planned to be released in Spark 4.1. The API is currently in heavy development and evolving. I think its safe to make this change at this point, given its still unreleased in a Spark Major version.

cc: @anishshri-db

yea not released yet. This is only in the 4.0 branch.
@sahnib - while we are here - is there a way to make this package private to sql too on the Java side ?

Java doesn't support package private with specifying ancestor package. It's always limited to the same package and it ~~won't~~ may not work for our case.
(It might change in recent versions though I haven't heard about such change.)

sahnib · 2024-06-03T15:05:23Z

Mind filing a JIRA please?

Filed SPARK-48511. Thanks.

sahnib · 2024-06-04T18:46:45Z

@HeartSaVioR PTAL, thanks.

HeartSaVioR · 2024-06-07T20:53:11Z

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

        watermarkPropagator.getInputWatermarkForEviction(tentativeBatchId,
          p.stateInfo.get.operatorId))
-    }.exists(_ == true)
+    }.contains(true)


nit: unrelated change? Please file a new minor PR.

will remove these minor changes from this PR.

HeartSaVioR

Thanks for the contribution. Left comments. General comments are 1) whether we can use any time mode where we could use None, or only ProcessingTime, and 2) why we need manual clock while we just removed out time mode None (doesn't sound to be relevant).

HeartSaVioR · 2024-06-07T20:55:34Z

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

 class ExpiredTimerInfoImpl(
    isValid: Boolean,
-    expiryTimeInMsOpt: Option[Long] = None,
-    timeMode: TimeMode = TimeMode.None()) extends ExpiredTimerInfo {


Do we assume we don't need to provide either it was from event time semantic vs processing time semantic? What was the rationale to add this and why this could be removed while we just remove out None?

That's a good question. TimeMode is passed in the StatefulProcessor.init function to the user. We were also passing it to the ExpiredTimerInfoImpl but the interface ExpiredTimerInfo does not even expose it to the user, and we never set this variable with proper timeMode. Hence, I have removed it. I dont think this was intended to here. cc: @anishshri-db can you confirm?

HeartSaVioR · 2024-06-07T20:56:09Z

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

    val lastExecutionRequiresAnotherBatch = noDataBatchesEnabled &&
      // need to check the execution plan of the previous batch
-      execCtx.previousContext.map { plan =>
+      execCtx.previousContext.exists { plan =>


nit: again, separate minor PR for unrelated change.

HeartSaVioR · 2024-06-07T20:57:43Z

sql/core/src/test/java/test/org/apache/spark/sql/JavaDatasetSuite.java

    Dataset<String> transformWithStateMapped = grouped.transformWithState(
      new TestStatefulProcessorWithInitialState(),
-      TimeMode.None(),
+      TimeMode.ProcessingTime(),


nit: just to be fully sure, either ProcessingTime or EventTime works, do I understand correctly? If either one only works for replacement of None, should be better to be documented.

either ProcessingTime or EventTime works as a replacement.

HeartSaVioR · 2024-06-07T20:58:37Z

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

      val store = provider.getStore(0)
      val handle = new StatefulProcessorHandleImpl(store, UUID.randomUUID(),
-        Encoders.STRING.asInstanceOf[ExpressionEncoder[Any]], TimeMode.None())
+        Encoders.STRING.asInstanceOf[ExpressionEncoder[Any]], TimeMode.ProcessingTime())


nit: same, wanted to know whether this is just a convenience or required.

just a convenience.

HeartSaVioR · 2024-06-07T21:02:09Z

sql/core/src/test/scala/org/apache/spark/sql/streaming/TransformWithListStateSuite.scala

I wonder why the test code change is required given the code change is just to remove TimeMode None. Do we fix some test flakiness as well here, or what is the rationale of manual clock?

TransformWithStateExec currently returns true for ProcessingTime timeMode in shouldRunAnotherBatch. (Note that this behavior is same as FMGWS). I had to add a manual clock to ensure that data processing is triggerred before Check/Expect blocks (otherwise AddData does not finish). Let me know if that answers your question.

HeartSaVioR · 2024-06-07T21:02:29Z

sql/core/src/test/scala/org/apache/spark/sql/streaming/TransformWithMapStateSuite.scala

Same question.

#46825 (comment)

HeartSaVioR · 2024-06-07T21:02:41Z

...core/src/test/scala/org/apache/spark/sql/streaming/TransformWithStateInitialStateSuite.scala

#46825 (comment)

HeartSaVioR · 2024-06-07T21:16:17Z

Also let's revisit the UX. If they use neither timeout nor TTL, they could just do None regardless they have event time column or not. Given we remove None, what is the expectation of UX? Does user have to specify either one based on what they have (event time column is set or not), even though they never use timeout? If that is the case, that doesn't sound like a better UX.

sahnib · 2024-06-12T15:47:07Z

Also let's revisit the UX. If they use neither timeout nor TTL, they could just do None regardless they have event time column or not. Given we remove None, what is the expectation of UX? Does user have to specify either one based on what they have (event time column is set or not), even though they never use timeout? If that is the case, that doesn't sound like a better UX.

Even with TimeMode None -> the query has eventTime (if eventTimeColumn is specified), and processingTime (based on driver's clock). I think this makes TimeMode None confusing.
As a general rule of thumb, customer should set TimeMode as EventTime if eventTimeColumn is specified, else use ProcessingTime.

HeartSaVioR · 2024-06-13T03:10:17Z

Just leaving a history here: currently processing time mode triggers the batch continuously which is not acceptable if there is no timer/TTL to check. We are now discussing how to handle the case.

github-actions · 2024-09-22T00:25:57Z

We're closing this PR because it hasn't been updated in a while. This isn't a judgement on the merit of the PR in any way. It's just a way of keeping the PR queue manageable.
If you'd like to revive this PR, please reopen it and ask a committer to remove the Stale tag!

github-actions bot added SQL STRUCTURED STREAMING labels May 31, 2024

sahnib force-pushed the remove-timeMode-None branch from 4981316 to b00a034 Compare May 31, 2024 21:12

sahnib marked this pull request as ready for review May 31, 2024 21:14

sahnib changed the title ~~Remove TimeMode None from TransformWithState.~~ [SS] Remove TimeMode None from TransformWithState. May 31, 2024

HyukjinKwon reviewed Jun 3, 2024

View reviewed changes

Remove TimeMode None from TransformWithState.

05ff15b

sahnib force-pushed the remove-timeMode-None branch from b00a034 to 05ff15b Compare June 3, 2024 15:02

sahnib changed the title ~~[SS] Remove TimeMode None from TransformWithState.~~ [SS][SPARK-48511] Remove TimeMode None from TransformWithState. Jun 3, 2024

anishshri-db approved these changes Jun 4, 2024

View reviewed changes

HeartSaVioR changed the title ~~[SS][SPARK-48511] Remove TimeMode None from TransformWithState.~~ [SPARK-48511][SS] Remove TimeMode None from TransformWithState. Jun 7, 2024

HeartSaVioR reviewed Jun 7, 2024

View reviewed changes

sahnib requested a review from HeartSaVioR June 12, 2024 15:38

github-actions bot added the Stale label Sep 22, 2024

github-actions bot closed this Sep 23, 2024

[SPARK-48511][SS] Remove TimeMode None from TransformWithState. #46825

[SPARK-48511][SS] Remove TimeMode None from TransformWithState. #46825

Uh oh!

Conversation

sahnib commented May 31, 2024 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

What changes were proposed in this pull request?

Why are the changes needed?

Does this PR introduce any user-facing change?

How was this patch tested?

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

Uh oh!

HyukjinKwon commented Jun 3, 2024

Uh oh!

Choose a reason for hiding this comment

Uh oh!

sahnib Jun 3, 2024 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

HeartSaVioR Jun 7, 2024 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

sahnib commented Jun 3, 2024

Uh oh!

sahnib commented Jun 4, 2024

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

HeartSaVioR left a comment • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

HeartSaVioR commented Jun 7, 2024 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Uh oh!

sahnib commented Jun 12, 2024

Uh oh!

HeartSaVioR commented Jun 13, 2024 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Uh oh!

github-actions bot commented Sep 22, 2024

Uh oh!

Reviewers

Assignees

Labels

Projects

sahnib commented May 31, 2024 •

edited

Loading

sahnib Jun 3, 2024 •

edited

Loading

HeartSaVioR Jun 7, 2024 •

edited

Loading

HeartSaVioR left a comment •

edited

Loading

HeartSaVioR commented Jun 7, 2024 •

edited

Loading

HeartSaVioR commented Jun 13, 2024 •

edited

Loading