[SPARK-53986][PYTHON][SS][TESTS] Reorganize Python streaming TWS test

### What changes were proposed in this pull request?

Reorganize the Python TWS tests:

- moving tws related tests to a new `/streaming` directory
- further split the Python TWS tests to smaller ones to speed up the CI

### Why are the changes needed?

Code refactoring

### Does this PR introduce _any_ user-facing change?

No

### How was this patch tested?

test is green

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

No

Closes #52691 from huanliwang-db/huanliwang-db/split-test-2.

Authored-by: huanliwang-db <[email protected]>
Signed-off-by: Ruifeng Zheng <[email protected]>

Author: huanliwang-db

dev/sparktestsupport/modules.py

@@ -662,8 +662,14 @@ def __hash__(self):
         "pyspark.sql.tests.streaming.test_streaming_foreach_batch",
         "pyspark.sql.tests.streaming.test_streaming_listener",
         "pyspark.sql.tests.pandas.test_pandas_grouped_map_with_state",
-        "pyspark.sql.tests.pandas.test_pandas_transform_with_state",
-        "pyspark.sql.tests.pandas.test_pandas_transform_with_state_checkpoint_v2",
+        "pyspark.sql.tests.pandas.streaming.test_pandas_transform_with_state",
+        "pyspark.sql.tests.pandas.streaming.test_pandas_transform_with_state_checkpoint_v2",
+        "pyspark.sql.tests.pandas.streaming.test_pandas_transform_with_state_state_variable",
+        "pyspark.sql.tests.pandas.streaming.test_pandas_transform_with_state_state_variable_checkpoint_v2",  # noqa: E501
+        "pyspark.sql.tests.pandas.streaming.test_transform_with_state",
+        "pyspark.sql.tests.pandas.streaming.test_transform_with_state_checkpoint_v2",
+        "pyspark.sql.tests.pandas.streaming.test_transform_with_state_state_variable",
+        "pyspark.sql.tests.pandas.streaming.test_transform_with_state_state_variable_checkpoint_v2",
     ],
     excluded_python_implementations=[
         "PyPy"  # Skip these tests under PyPy since they require numpy and it isn't available there
@@ -1182,9 +1188,11 @@ def __hash__(self):
         "pyspark.sql.tests.connect.streaming.test_parity_listener",
         "pyspark.sql.tests.connect.streaming.test_parity_foreach",
         "pyspark.sql.tests.connect.streaming.test_parity_foreach_batch",
-        "pyspark.sql.tests.connect.streaming.test_parity_transform_with_state_pyspark",
-        "pyspark.sql.tests.connect.pandas.test_parity_pandas_grouped_map_with_state",
-        "pyspark.sql.tests.connect.pandas.test_parity_pandas_transform_with_state",
+        "pyspark.sql.tests.connect.pandas.streaming.test_parity_pandas_grouped_map_with_state",
+        "pyspark.sql.tests.connect.pandas.streaming.test_parity_pandas_transform_with_state",
+        "pyspark.sql.tests.connect.pandas.streaming.test_parity_pandas_transform_with_state_state_variable",  # noqa: E501
+        "pyspark.sql.tests.connect.pandas.streaming.test_parity_transform_with_state",
+        "pyspark.sql.tests.connect.pandas.streaming.test_parity_transform_with_state_state_variable",  # noqa: E501
     ],
     excluded_python_implementations=[
         "PyPy"  # Skip these tests under PyPy since they require numpy and it isn't available there

python/pyspark/sql/tests/connect/pandas/streaming/__init__.py

@@ -0,0 +1,16 @@
+#
+# Licensed to the Apache Software Foundation (ASF) under one or more
+# contributor license agreements.  See the NOTICE file distributed with
+# this work for additional information regarding copyright ownership.
+# The ASF licenses this file to You under the Apache License, Version 2.0
+# (the "License"); you may not use this file except in compliance with
+# the License.  You may obtain a copy of the License at
+#
+#    http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+#

python/pyspark/sql/tests/connect/pandas/test_parity_pandas_grouped_map_with_state.py renamed to python/pyspark/sql/tests/connect/pandas/streaming/test_parity_pandas_grouped_map_with_state.py

@@ -29,7 +29,7 @@ class GroupedApplyInPandasWithStateTests(
 
 
 if __name__ == "__main__":
-    from pyspark.sql.tests.connect.pandas.test_parity_pandas_grouped_map_with_state import *  # noqa: F401,E501
+    from pyspark.sql.tests.connect.pandas.streaming.test_parity_pandas_grouped_map_with_state import *  # noqa: F401,E501
 
     try:
         import xmlrunner

python/pyspark/sql/tests/connect/pandas/test_parity_pandas_transform_with_state.py renamed to python/pyspark/sql/tests/connect/pandas/streaming/test_parity_pandas_transform_with_state.py

@@ -16,7 +16,7 @@
 #
 import unittest
 
-from pyspark.sql.tests.pandas.test_pandas_transform_with_state import (
+from pyspark.sql.tests.pandas.streaming.test_pandas_transform_with_state import (
     TransformWithStateInPandasTestsMixin,
 )
 from pyspark import SparkConf
@@ -54,7 +54,7 @@ def test_schema_evolution_scenarios(self):
 
 
 if __name__ == "__main__":
-    from pyspark.sql.tests.connect.pandas.test_parity_pandas_transform_with_state import *  # noqa: F401,E501
+    from pyspark.sql.tests.connect.pandas.streaming.test_parity_pandas_transform_with_state import *  # noqa: F401,E501
 
     try:
         import xmlrunner

python/pyspark/sql/tests/connect/pandas/streaming/test_parity_pandas_transform_with_state_state_variable.py

@@ -0,0 +1,61 @@
+#
+# Licensed to the Apache Software Foundation (ASF) under one or more
+# contributor license agreements.  See the NOTICE file distributed with
+# this work for additional information regarding copyright ownership.
+# The ASF licenses this file to You under the Apache License, Version 2.0
+# (the "License"); you may not use this file except in compliance with
+# the License.  You may obtain a copy of the License at
+#
+#    http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+#
+import unittest
+
+from pyspark.sql.tests.pandas.streaming.test_pandas_transform_with_state_state_variable import (
+    TransformWithStateInPandasStateVariableTestsMixin,
+)
+from pyspark import SparkConf
+from pyspark.testing.connectutils import ReusedConnectTestCase
+
+
+class TransformWithStateInPandasStateVariableParityTests(
+    TransformWithStateInPandasStateVariableTestsMixin, ReusedConnectTestCase
+):
+    """
+    Spark connect parity tests for TransformWithStateInPandas. Run every test case in
+     `TransformWithStateInPandasStateVariableTestsMixin` in spark connect mode.
+    """
+
+    @classmethod
+    def conf(cls):
+        # Due to multiple inheritance from the same level, we need to explicitly setting configs in
+        # both TransformWithStateInPandasStateVariableTestsMixin and ReusedConnectTestCase here
+        cfg = SparkConf(loadDefaults=False)
+        for base in cls.__bases__:
+            if hasattr(base, "conf"):
+                parent_cfg = base.conf()
+                for k, v in parent_cfg.getAll():
+                    cfg.set(k, v)
+
+        # Extra removing config for connect suites
+        if cfg._jconf is not None:
+            cfg._jconf.remove("spark.master")
+
+        return cfg
+
+
+if __name__ == "__main__":
+    from pyspark.sql.tests.connect.pandas.streaming.test_parity_pandas_transform_with_state_state_variable import *  # noqa: F401,E501
+
+    try:
+        import xmlrunner
+
+        testRunner = xmlrunner.XMLTestRunner(output="target/test-reports", verbosity=2)
+    except ImportError:
+        testRunner = None
+    unittest.main(testRunner=testRunner, verbosity=2)

python/pyspark/sql/tests/connect/streaming/test_parity_transform_with_state_pyspark.py renamed to python/pyspark/sql/tests/connect/pandas/streaming/test_parity_transform_with_state.py

@@ -16,7 +16,7 @@
 #
 import unittest
 
-from pyspark.sql.tests.pandas.test_pandas_transform_with_state import (
+from pyspark.sql.tests.pandas.streaming.test_transform_with_state import (
     TransformWithStateInPySparkTestsMixin,
 )
 from pyspark import SparkConf
@@ -54,7 +54,7 @@ def test_schema_evolution_scenarios(self):
 
 
 if __name__ == "__main__":
-    from pyspark.sql.tests.connect.streaming.test_parity_transform_with_state_pyspark import *  # noqa: F401,E501
+    from pyspark.sql.tests.connect.pandas.streaming.test_parity_transform_with_state import *  # noqa: F401,E501
 
     try:
         import xmlrunner

python/pyspark/sql/tests/connect/pandas/streaming/test_parity_transform_with_state_state_variable.py

@@ -0,0 +1,61 @@
+#
+# Licensed to the Apache Software Foundation (ASF) under one or more
+# contributor license agreements.  See the NOTICE file distributed with
+# this work for additional information regarding copyright ownership.
+# The ASF licenses this file to You under the Apache License, Version 2.0
+# (the "License"); you may not use this file except in compliance with
+# the License.  You may obtain a copy of the License at
+#
+#    http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+#
+import unittest
+
+from pyspark.sql.tests.pandas.streaming.test_transform_with_state_state_variable import (
+    TransformWithStateInPySparkStateVariableTestsMixin,
+)
+from pyspark import SparkConf
+from pyspark.testing.connectutils import ReusedConnectTestCase
+
+
+class TransformWithStateInPySparkStateVariableParityTests(
+    TransformWithStateInPySparkStateVariableTestsMixin, ReusedConnectTestCase
+):
+    """
+    Spark connect parity tests for TransformWithStateInPySparkStateVariable. Run every test case in
+     `TransformWithStateInPySparkStateVariableTestsMixin` in spark connect mode.
+    """
+
+    @classmethod
+    def conf(cls):
+        # Due to multiple inheritance from the same level, we need to explicitly setting configs in
+        # both TransformWithStateInPySparkStateVariableTestsMixin and ReusedConnectTestCase here
+        cfg = SparkConf(loadDefaults=False)
+        for base in cls.__bases__:
+            if hasattr(base, "conf"):
+                parent_cfg = base.conf()
+                for k, v in parent_cfg.getAll():
+                    cfg.set(k, v)
+
+        # Extra removing config for connect suites
+        if cfg._jconf is not None:
+            cfg._jconf.remove("spark.master")
+
+        return cfg
+
+
+if __name__ == "__main__":
+    from pyspark.sql.tests.connect.pandas.streaming.test_parity_transform_with_state_state_variable import *  # noqa: F401,E501
+
+    try:
+        import xmlrunner
+
+        testRunner = xmlrunner.XMLTestRunner(output="target/test-reports", verbosity=2)
+    except ImportError:
+        testRunner = None
+    unittest.main(testRunner=testRunner, verbosity=2)

python/pyspark/sql/tests/pandas/streaming/__init__.py

@@ -0,0 +1,16 @@
+#
+# Licensed to the Apache Software Foundation (ASF) under one or more
+# contributor license agreements.  See the NOTICE file distributed with
+# this work for additional information regarding copyright ownership.
+# The ASF licenses this file to You under the Apache License, Version 2.0
+# (the "License"); you may not use this file except in compliance with
+# the License.  You may obtain a copy of the License at
+#
+#    http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+#