Revert "temp skip tests"

This reverts commit 898561d.

Author: jiateoh

python/pyspark/sql/tests/pandas/test_pandas_transform_with_state.py

@@ -194,7 +194,6 @@ def _test_transform_with_state_basic(
         q.awaitTermination(10)
         self.assertTrue(q.exception() is None)
 
-    @unittest.skip("Temporarily disabled for testing")
     def test_transform_with_state_basic(self):
         def check_results(batch_df, batch_id):
             batch_df.collect()
@@ -211,7 +210,6 @@ def check_results(batch_df, batch_id):
 
         self._test_transform_with_state_basic(SimpleStatefulProcessorFactory(), check_results)
 
-    @unittest.skip("Temporarily disabled for testing")
     def test_transform_with_state_non_exist_value_state(self):
         def check_results(batch_df, _):
             batch_df.collect()
@@ -224,7 +222,6 @@ def check_results(batch_df, _):
             InvalidSimpleStatefulProcessorFactory(), check_results, True
         )
 
-    @unittest.skip("Temporarily disabled for testing")
     def test_transform_with_state_query_restarts(self):
         root_path = tempfile.mkdtemp()
         input_path = root_path + "/input"
@@ -299,7 +296,6 @@ def test_transform_with_state_query_restarts(self):
             Row(id="1", countAsString="2"),
         }
 
-    @unittest.skip("Temporarily disabled for testing")
     def test_transform_with_state_list_state(self):
         def check_results(batch_df, _):
             batch_df.collect()
@@ -312,7 +308,6 @@ def check_results(batch_df, _):
             ListStateProcessorFactory(), check_results, True, "processingTime"
         )
 
-    @unittest.skip("Temporarily disabled for testing")
     def test_transform_with_state_list_state_large_list(self):
         def check_results(batch_df, batch_id):
             batch_df.collect()
@@ -388,7 +383,6 @@ def check_results(batch_df, batch_id):
         self.assertTrue(q.exception() is None)
 
     # test list state with ttl has the same behavior as list state when state doesn't expire.
-    @unittest.skip("Temporarily disabled for testing")
     def test_transform_with_state_list_state_large_ttl(self):
         def check_results(batch_df, batch_id):
             batch_df.collect()
@@ -401,7 +395,6 @@ def check_results(batch_df, batch_id):
             ListStateLargeTTLProcessorFactory(), check_results, True, "processingTime"
         )
 
-    @unittest.skip("Temporarily disabled for testing")
     def test_transform_with_state_map_state(self):
         def check_results(batch_df, _):
             batch_df.collect()
@@ -413,7 +406,6 @@ def check_results(batch_df, _):
         self._test_transform_with_state_basic(MapStateProcessorFactory(), check_results, True)
 
     # test map state with ttl has the same behavior as map state when state doesn't expire.
-    @unittest.skip("Temporarily disabled for testing")
     def test_transform_with_state_map_state_large_ttl(self):
         def check_results(batch_df, batch_id):
             batch_df.collect()
@@ -428,7 +420,6 @@ def check_results(batch_df, batch_id):
 
     # test value state with ttl has the same behavior as value state when
     # state doesn't expire.
-    @unittest.skip("Temporarily disabled for testing")
     def test_value_state_ttl_basic(self):
         def check_results(batch_df, batch_id):
             batch_df.collect()
@@ -449,7 +440,6 @@ def check_results(batch_df, batch_id):
 
     # TODO SPARK-50908 holistic fix for TTL suite
     @unittest.skip("test is flaky and it is only a timing issue, skipping until we can resolve")
-    @unittest.skip("Temporarily disabled for testing")
     def test_value_state_ttl_expiration(self):
         def check_results(batch_df, batch_id):
             batch_df.collect()
@@ -599,7 +589,6 @@ def _test_transform_with_state_proc_timer(self, stateful_processor_factory, chec
         q.awaitTermination(10)
         self.assertTrue(q.exception() is None)
 
-    @unittest.skip("Temporarily disabled for testing")
     def test_transform_with_state_proc_timer(self):
         def check_results(batch_df, batch_id):
             batch_df.collect()
@@ -717,7 +706,6 @@ def prepare_batch3(input_path):
         q.awaitTermination(10)
         self.assertTrue(q.exception() is None)
 
-    @unittest.skip("Temporarily disabled for testing")
     def test_transform_with_state_event_time(self):
         def check_results(batch_df, batch_id):
             batch_df.collect()
@@ -750,7 +738,6 @@ def check_results(batch_df, batch_id):
             EventTimeStatefulProcessorFactory(), check_results
         )
 
-    @unittest.skip("Temporarily disabled for testing")
     def test_transform_with_state_with_wmark_and_non_event_time(self):
         def check_results(batch_df, batch_id):
             batch_df.collect()
@@ -849,7 +836,6 @@ def _test_transform_with_state_init_state(
         q.awaitTermination(10)
         self.assertTrue(q.exception() is None)
 
-    @unittest.skip("Temporarily disabled for testing")
     def test_transform_with_state_init_state(self):
         def check_results(batch_df, batch_id):
             batch_df.collect()
@@ -874,7 +860,6 @@ def check_results(batch_df, batch_id):
             SimpleStatefulProcessorWithInitialStateFactory(), check_results
         )
 
-    @unittest.skip("Temporarily disabled for testing")
     def test_transform_with_state_init_state_with_extra_transformation(self):
         def check_results(batch_df, batch_id):
             batch_df.collect()
@@ -954,7 +939,6 @@ def _test_transform_with_state_non_contiguous_grouping_cols(
         q.awaitTermination(10)
         self.assertTrue(q.exception() is None)
 
-    @unittest.skip("Temporarily disabled for testing")
     def test_transform_with_state_non_contiguous_grouping_cols(self):
         def check_results(batch_df, batch_id):
             batch_df.collect()
@@ -967,7 +951,6 @@ def check_results(batch_df, batch_id):
             SimpleStatefulProcessorWithInitialStateFactory(), check_results
         )
 
-    @unittest.skip("Temporarily disabled for testing")
     def test_transform_with_state_non_contiguous_grouping_cols_with_init_state(self):
         def check_results(batch_df, batch_id):
             batch_df.collect()
@@ -1051,7 +1034,6 @@ def _test_transform_with_state_chaining_ops(
         q.processAllAvailable()
         q.awaitTermination(10)
 
-    @unittest.skip("Temporarily disabled for testing")
     def test_transform_with_state_chaining_ops(self):
         def check_results(batch_df, batch_id):
             batch_df.collect()
@@ -1088,7 +1070,6 @@ def check_results(batch_df, batch_id):
             ["outputTimestamp", "id"],
         )
 
-    @unittest.skip("Temporarily disabled for testing")
     def test_transform_with_state_init_state_with_timers(self):
         def check_results(batch_df, batch_id):
             batch_df.collect()
@@ -1119,7 +1100,6 @@ def check_results(batch_df, batch_id):
             StatefulProcessorWithInitialStateTimersFactory(), check_results, "processingTime"
         )
 
-    @unittest.skip("Temporarily disabled for testing")
     def test_transform_with_state_batch_query(self):
         data = [("0", 123), ("0", 46), ("1", 146), ("1", 346)]
         df = self.spark.createDataFrame(data, "id string, temperature int")
@@ -1151,7 +1131,6 @@ def test_transform_with_state_batch_query(self):
             Row(id="1", countAsString="2"),
         }
 
-    @unittest.skip("Temporarily disabled for testing")
     def test_transform_with_state_batch_query_initial_state(self):
         data = [("0", 123), ("0", 46), ("1", 146), ("1", 346)]
         df = self.spark.createDataFrame(data, "id string, temperature int")
@@ -1196,11 +1175,9 @@ def test_transform_with_state_batch_query_initial_state(self):
     @unittest.skipIf(
         "COVERAGE_PROCESS_START" in os.environ, "Flaky with coverage enabled, skipping for now."
     )
-    @unittest.skip("Temporarily disabled for testing")
     def test_transform_with_map_state_metadata(self):
         self._test_transform_with_map_state_metadata(None)
 
-    @unittest.skip("Temporarily disabled for testing")
     def test_transform_with_map_state_metadata_with_init_state(self):
         # run the same test suite again but with no-op initial state
         # TWS with initial state is using a different python runner
@@ -1333,7 +1310,6 @@ def check_results(batch_df, batch_id):
         )
 
     # This test covers multiple list state variables and flatten option
-    @unittest.skip("Temporarily disabled for testing")
     def test_transform_with_list_state_metadata(self):
         checkpoint_path = tempfile.mktemp()
 
@@ -1414,7 +1390,6 @@ def check_results(batch_df, batch_id):
 
     # This test covers value state variable and read change feed,
     # snapshotStartBatchId related options
-    @unittest.skip("Temporarily disabled for testing")
     def test_transform_with_value_state_metadata(self):
         checkpoint_path = tempfile.mktemp()
 
@@ -1505,7 +1480,6 @@ def check_results(batch_df, batch_id):
                 checkpoint_path=checkpoint_path,
             )
 
-    @unittest.skip("Temporarily disabled for testing")
     def test_transform_with_state_restart_with_multiple_rows_init_state(self):
         def check_results(batch_df, _):
             batch_df.collect()
@@ -1565,7 +1539,6 @@ def dataframe_to_value_list(output_df):
             initial_state=init_df,
         )
 
-    @unittest.skip("Temporarily disabled for testing")
     def test_transform_with_state_in_pandas_composite_type(self):
         def check_results(batch_df, batch_id):
             if batch_id == 0:
@@ -1624,7 +1597,6 @@ def check_results(batch_df, batch_id):
         )
 
     # run the same test suites again but with single shuffle partition
-    @unittest.skip("Temporarily disabled for testing")
     def test_transform_with_state_with_timers_single_partition(self):
         with self.sql_conf({"spark.sql.shuffle.partitions": "1"}):
             self.test_transform_with_state_init_state_with_timers()
@@ -1673,7 +1645,6 @@ def _run_evolution_test(self, processor_factory, checkpoint_dir, check_results,
         q.processAllAvailable()
         q.awaitTermination(10)
 
-    @unittest.skip("Temporarily disabled for testing")
     def test_schema_evolution_scenarios(self):
         """Test various schema evolution scenarios"""
         with self.sql_conf({"spark.sql.streaming.stateStore.encodingFormat": "avro"}):
@@ -1742,7 +1713,6 @@ def check_upcast(batch_df, batch_id):
 
     # This test case verifies that an exception is thrown when downcasting, which violates
     # Avro's schema evolution rules
-    @unittest.skip("Temporarily disabled for testing")
     def test_schema_evolution_fails(self):
         with self.sql_conf({"spark.sql.streaming.stateStore.encodingFormat": "avro"}):
             with tempfile.TemporaryDirectory() as checkpoint_dir:
@@ -1795,7 +1765,6 @@ def check_basic_state(batch_df, batch_id):
                         and "Schema evolution is not possible" in error_msg
                     )
 
-    @unittest.skip("Temporarily disabled for testing")
     def test_not_nullable_fails(self):
         with self.sql_conf({"spark.sql.streaming.stateStore.encodingFormat": "avro"}):
             with tempfile.TemporaryDirectory() as checkpoint_dir:
@@ -1828,7 +1797,6 @@ def check_basic_state(batch_df, batch_id):
                         and "column family state must be nullable" in error_msg
                     )
 
-    @unittest.skip("Temporarily disabled for testing")
     def test_transform_with_state_int_to_decimal_coercion(self):
         if not self.use_pandas():
             return