bogao007 commented on code in PR #41558:
URL: https://github.com/apache/spark/pull/41558#discussion_r1227320572
##########
connector/connect/client/jvm/src/test/scala/org/apache/spark/sql/KeyValueGroupedDatasetE2ETestSuite.scala:
##########
@@ -447,4 +449,20 @@ class KeyValueGroupedDatasetE2ETestSuite extends QueryTest
with SQLHelper {
checkDataset(keys, "1", "2", "10", "20")
}
+
+ test("flatMapGroupsWithState") {
+ val stateFunc = (key: String, values: Iterator[String], state:
GroupState[Int]) => {
+ if (state.exists) throw new IllegalArgumentException("state.exists
should be false")
+ Iterator((key, values.size))
+ }
+
+ val session: SparkSession = spark
+ import session.implicits._
+ val values = Seq("a", "a", "b", "c", "c", "c", "c").toDS()
+ .groupByKey(x => x)
+ .flatMapGroupsWithState(Append, GroupStateTimeout.NoTimeout)(stateFunc)
Review Comment:
When I tried to add `mapGroupsWithState` test, it actually failed with below
error when I call `.collect()`.
```
io.grpc.StatusRuntimeException: INTERNAL: Job aborted due to stage failure:
Task 0 in stage 112.0 failed 1 times, most recent failure: Lost task 0.0 in
stage 112.0 (TID 270) (192.168.2.82 executor driver):
java.lang.ClassCastException: class scala.Tuple2 cannot be cast to class
scala.collection.TraversableOnce (scala.Tuple2 and
scala.collection.TraversableOnce are in unnamed module of loader 'app')
```
Do you have any insights on this? Not sure if it's an issue with my
implementation or tests.
--
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]
