LuciferYang commented on code in PR #49277:
URL: https://github.com/apache/spark/pull/49277#discussion_r1977336954
##########
python/pyspark/sql/tests/pandas/test_pandas_transform_with_state.py:
##########
@@ -1294,6 +1310,167 @@ def
test_transform_with_state_with_timers_single_partition(self):
self.test_transform_with_state_in_pandas_proc_timer()
self.test_transform_with_state_restart_with_multiple_rows_init_state()
+ def _run_evolution_test(
+ self, processor, checkpoint_dir, check_results, df,
check_exception=None
+ ):
+ output_schema = StructType(
+ [
+ StructField("id", StringType(), True),
+ StructField("value", processor.state_schema, True),
+ ]
+ )
+
+ # Stop any active streams first
+ for q in self.spark.streams.active:
+ q.stop()
+
+ try:
+ q = (
+ df.groupBy("id")
+ .transformWithStateInPandas(
+ statefulProcessor=processor,
+ outputStructType=output_schema,
+ outputMode="Update",
+ timeMode="None",
+ )
+ .writeStream.queryName("evolution_test")
+ .option("checkpointLocation", checkpoint_dir)
+ .foreachBatch(check_results)
+ .outputMode("update")
+ .start()
+ )
+
+ self.assertEqual(q.name, "evolution_test")
+ self.assertTrue(q.isActive)
+ q.processAllAvailable()
+ q.awaitTermination(10)
+
+ if q.exception() is None:
+ assert check_exception is None
+
+ except Exception as e:
+ # If we are expecting an exception, verify it's the right one
+ if check_exception is None:
+ raise # Re-raise if we weren't expecting an exception
+ self.assertTrue(check_exception(e))
+
+ def test_schema_evolution_scenarios(self):
Review Comment:
Is this a test-only issue or a connect issue? Will this block the release of
Apache Spark 4.0?
--
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: reviews-unsubscr...@spark.apache.org
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscr...@spark.apache.org
For additional commands, e-mail: reviews-h...@spark.apache.org
