temp skip tests

Author: jiateoh

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

@@ -194,6 +194,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()
@@ -210,6 +211,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()
@@ -222,6 +224,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"
@@ -296,6 +299,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()
@@ -308,6 +312,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()
@@ -383,6 +388,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()
@@ -395,6 +401,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()
@@ -406,6 +413,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()
@@ -420,6 +428,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()
@@ -440,6 +449,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()
@@ -589,6 +599,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()
@@ -706,6 +717,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()
@@ -738,6 +750,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()
@@ -836,6 +849,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()
@@ -860,6 +874,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()
@@ -939,6 +954,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()
@@ -951,6 +967,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()
@@ -1034,6 +1051,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()
@@ -1070,6 +1088,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()
@@ -1100,6 +1119,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")
@@ -1131,6 +1151,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")
@@ -1175,9 +1196,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
@@ -1310,6 +1333,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()
 
@@ -1390,6 +1414,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()
 
@@ -1480,6 +1505,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()
@@ -1539,6 +1565,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:
@@ -1597,6 +1624,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()
@@ -1645,6 +1673,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"}):
@@ -1713,6 +1742,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:
@@ -1765,6 +1795,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:
@@ -1797,6 +1828,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