temp skip tests

Author: jiateoh

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

@@ -200,6 +200,7 @@ 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()
@@ -216,6 +217,7 @@ 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()
@@ -228,6 +230,7 @@ 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"
@@ -302,6 +305,7 @@ 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()
@@ -314,6 +318,7 @@ 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()
@@ -389,6 +394,7 @@ 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,6 +407,7 @@ 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()
@@ -412,6 +419,7 @@ 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()
@@ -426,6 +434,7 @@ 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()
@@ -446,6 +455,7 @@ 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()
@@ -595,6 +605,7 @@ 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()
@@ -712,6 +723,7 @@ 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()
@@ -744,6 +756,7 @@ 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()
@@ -842,6 +855,7 @@ 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()
@@ -866,6 +880,7 @@ 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()
@@ -945,6 +960,7 @@ 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()
@@ -957,6 +973,7 @@ 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()
@@ -1040,6 +1057,7 @@ 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()
@@ -1076,6 +1094,7 @@ 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()
@@ -1106,6 +1125,7 @@ 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")
@@ -1137,6 +1157,7 @@ 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")
@@ -1181,9 +1202,11 @@ 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
@@ -1316,6 +1339,7 @@ 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()
 
@@ -1396,6 +1420,7 @@ 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()
 
@@ -1486,6 +1511,7 @@ 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()
@@ -1545,6 +1571,7 @@ 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:
@@ -1904,6 +1931,7 @@ 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()
@@ -1952,6 +1980,7 @@ 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"}):
@@ -2020,6 +2049,7 @@ 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:
@@ -2072,6 +2102,7 @@ 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:
@@ -2104,6 +2135,7 @@ 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