[SPARK-51869][SS] Create classification for user errors within UDFs for Scala TransformWithState #50667

ericm-db · 2025-04-22T21:54:06Z

What changes were proposed in this pull request?

This PR adds proper error handling to the handleInputRows method in TransformWithState

Why are the changes needed?

The changes improve error handling and reporting when users implement custom stateful processors. Without this change, when an error occurs within the handleInputRows function, users would receive a generic exception without clear indication that the error originated from their implementation. The new approach provides more descriptive error messages that pinpoint the source of the problem.

Does this PR introduce any user-facing change?

Yes, this PR introduces a user-facing change in the form of a new error message format for errors occurring within the StatefulProcessor's handleInputRows function. Users will now receive more descriptive error messages when their stateful processing logic fails.

How was this patch tested?

The patch was tested by adding two new test cases to TransformWithStateSuite:

"transformWithState - check that error within handleInputRows is classified" - tests that unclassified errors within the handleInputRows method are properly wrapped in a TransformWithStateUserFunctionException
"transformWithState - check that classified error is thrown from handleInputRows" - tests that already classified SparkThrowable exceptions are passed through correctly

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

No

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

common/utils/src/main/resources/error/error-conditions.json

anishshri-db · 2025-04-23T05:50:17Z

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

+      }
+      ImplicitGroupingKeyTracker.removeImplicitKey()
+    } catch {
+      case sparkThrowable: SparkThrowable =>


We need to check if the underlying exception has the error class already defined ?

Is the existing code not sufficient? This is how ForeachBatch does it:

try { batchWriter(ds, batchId) } catch { // The user code can throw any type of exception. case NonFatal(e) if !e.isInstanceOf[SparkThrowable] => throw ForeachBatchUserFuncException(e) }

Can we also check for error class explicitly also - just to be safe ?

Ohh sure yeah sorry - didn't realize that's what you meant. Yeah will add.

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

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

anishshri-db

lgtm pending minor nits

ericm-db · 2025-04-23T06:19:29Z

@HeartSaVioR can you PTAL when you get a chance?

HeartSaVioR

Looks good overall, only one question for clarifying.

HeartSaVioR · 2025-04-27T22:20:01Z

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

-      getOutputRow(obj)
-    }
-    ImplicitGroupingKeyTracker.removeImplicitKey()
+    withStatefulProcessorErrorHandling("handleInputRows") {


Do we have specific reason to wrap broader scope of code instead of strictly the scope of user function call?

Ah good point

HeartSaVioR · 2025-04-27T22:20:09Z

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

-        .handleInitialState(keyObj, initState,
-          new TimerValuesImpl(batchTimestampMs, eventTimeWatermarkForEviction))
+    withStatefulProcessorErrorHandling("handleInitialState") {
+      val getInitStateValueObj =


HeartSaVioR · 2025-04-27T22:20:18Z

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

-      getOutputRow(obj)
-    }
-    ImplicitGroupingKeyTracker.removeImplicitKey()
+    withStatefulProcessorErrorHandling("handleExpiredTimer") {


HeartSaVioR

+1 pending CI

HeartSaVioR · 2025-04-28T05:28:57Z

This only failed with docker integration CI.
https://github.com/ericm-db/spark/runs/41243071103

HeartSaVioR · 2025-04-28T05:29:03Z

Thanks! Merging to master.

…or Scala TransformWithState ### What changes were proposed in this pull request? This PR adds proper error handling to the handleInputRows method in TransformWithState ### Why are the changes needed? The changes improve error handling and reporting when users implement custom stateful processors. Without this change, when an error occurs within the handleInputRows function, users would receive a generic exception without clear indication that the error originated from their implementation. The new approach provides more descriptive error messages that pinpoint the source of the problem. ### Does this PR introduce any user-facing change? Yes, this PR introduces a user-facing change in the form of a new error message format for errors occurring within the StatefulProcessor's handleInputRows function. Users will now receive more descriptive error messages when their stateful processing logic fails. ### How was this patch tested? The patch was tested by adding two new test cases to TransformWithStateSuite: "transformWithState - check that error within handleInputRows is classified" - tests that unclassified errors within the handleInputRows method are properly wrapped in a TransformWithStateUserFunctionException "transformWithState - check that classified error is thrown from handleInputRows" - tests that already classified SparkThrowable exceptions are passed through correctly ### Was this patch authored or co-authored using generative AI tooling? No Closes apache#50667 from ericm-db/tws-user-error. Authored-by: Eric Marnadi <[email protected]> Signed-off-by: Jungtaek Lim <[email protected]>

github-actions bot added SQL STRUCTURED STREAMING PYTHON labels Apr 22, 2025

ericm-db force-pushed the tws-user-error branch from 4f5005c to 5db1f4e Compare April 22, 2025 22:48

github-actions bot removed the PYTHON label Apr 22, 2025

ericm-db changed the title ~~[WIP] Creating user error classification for TransformWithState~~ [SPARK-51869] Create classification for user errors within handleInputRows for Scala TransformWithState Apr 22, 2025

HyukjinKwon changed the title ~~[SPARK-51869] Create classification for user errors within handleInputRows for Scala TransformWithState~~ [SPARK-51869][SS] Create classification for user errors within handleInputRows for Scala TransformWithState Apr 22, 2025

anishshri-db reviewed Apr 22, 2025

View reviewed changes

sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/TransformWithStateExec.scala Outdated Show resolved Hide resolved

anishshri-db reviewed Apr 23, 2025

View reviewed changes

common/utils/src/main/resources/error/error-conditions.json Outdated Show resolved Hide resolved

ericm-db added 5 commits April 22, 2025 22:22

[WIP] Creating user error classification for TransformWithState

91541a3

adding error handling for other udf

4eff6dd

adding tests for handleInitialState, handleExpiredTimers

4c6a10f

for close too

8486459

adding func as param

3f17b22

ericm-db force-pushed the tws-user-error branch from 46aa99c to 3f17b22 Compare April 23, 2025 05:23

anishshri-db reviewed Apr 23, 2025

View reviewed changes

common/utils/src/main/resources/error/error-conditions.json Outdated Show resolved Hide resolved

error message

c724556

anishshri-db reviewed Apr 23, 2025

View reviewed changes

period

fcf4b96

anishshri-db reviewed Apr 23, 2025

View reviewed changes

sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/TransformWithStateExec.scala Outdated Show resolved Hide resolved

ericm-db added 2 commits April 22, 2025 22:58

refactoring

f440bcd

checking if error condition is null

ee86331

anishshri-db reviewed Apr 23, 2025

View reviewed changes

sql/core/src/test/scala/org/apache/spark/sql/streaming/TransformWithStateSuite.scala Outdated Show resolved Hide resolved

anishshri-db reviewed Apr 23, 2025

View reviewed changes

sql/core/src/test/scala/org/apache/spark/sql/streaming/TransformWithStateSuite.scala Show resolved Hide resolved

anishshri-db reviewed Apr 23, 2025

View reviewed changes

sql/core/src/test/scala/org/apache/spark/sql/streaming/TransformWithStateSuite.scala Outdated Show resolved Hide resolved

anishshri-db reviewed Apr 23, 2025

View reviewed changes

sql/core/src/test/scala/org/apache/spark/sql/streaming/TransformWithStateSuite.scala Outdated Show resolved Hide resolved

anishshri-db reviewed Apr 23, 2025

View reviewed changes

sql/core/src/test/scala/org/apache/spark/sql/streaming/TransformWithStateSuite.scala Outdated Show resolved Hide resolved

correcting tests

fd9fc6b

anishshri-db approved these changes Apr 23, 2025

View reviewed changes

added dummy exception

d66a88b

ericm-db changed the title ~~[SPARK-51869][SS] Create classification for user errors within handleInputRows for Scala TransformWithState~~ [SPARK-51869][SS] Create classification for user errors within UDFs for Scala TransformWithState Apr 23, 2025

HeartSaVioR reviewed Apr 27, 2025

View reviewed changes

ericm-db added 2 commits April 27, 2025 18:17

indents

1a9eafa

adding space back

a338409

HeartSaVioR approved these changes Apr 28, 2025

View reviewed changes

HeartSaVioR closed this in b83a819 Apr 28, 2025

[SPARK-51869][SS] Create classification for user errors within UDFs for Scala TransformWithState #50667

[SPARK-51869][SS] Create classification for user errors within UDFs for Scala TransformWithState #50667

Uh oh!

Conversation

ericm-db commented Apr 22, 2025 • 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!

Uh oh!

Uh oh!

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!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

anishshri-db left a comment

Choose a reason for hiding this comment

Uh oh!

ericm-db commented Apr 23, 2025

Uh oh!

HeartSaVioR left a comment

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 left a comment

Choose a reason for hiding this comment

Uh oh!

HeartSaVioR commented Apr 28, 2025

Uh oh!

HeartSaVioR commented Apr 28, 2025

Uh oh!

Uh oh!

ericm-db commented Apr 22, 2025 •

edited

Loading