Enable prism by default (where supported) (#34612)

* Enable prism by default

* Clean up fallback code

* Clean up fallback code

* Add missing base case

* Add missing base case

* Fix tests

* Fix tests

* exclude unsupported state

* Exclude locally materialized results

* Fix fallback/dupe execution

* Fix some snippet tests

* Disable argparse abbreviation

* Some test fixes for prism switch

* lint

* row fixes

* Make row comp logic default

* Add more yaml examples

* Fix dataframes tests

* Examples fixes

* type hints

* Some more transform test fixes

* more test fixes

* More generic error checks

* Reshuffle tests

* enrichment error catching

* more test fixes

* Scope out external transforms

* simplify test

* Fix more tests

* Clean up test

* correct runner mode

* ML tests

* ib collect test

* Make sure assertions dont fire in incorrect order

* ML test fixes

* typing

* More fixes

* some more fixes

* Another error fix

* Temporarily set log level to debug

* yapf

* More error regex fixes

* Fix some error messages/metric tests

* more generic tests

* Upgrade logging to warning to see what is happening

* Some more patches

* Wait until finish for test pipelines

* fix test_always

* A few more small fixes

* Temporarily update logging

* clean up merge

* Some more exclusions

* regex isssues

* Propogate original failure when using instruction cache

* Batching fix

* Trigger some postcommits

* fmt

* More test fixes

* Fix a few more tests

* linting

* Bump workflow timeout (#35420)

* Bump workflow timeout

* Update beam_PostCommit_Python_Xlang_Gcp_Direct.yml

* linting/fixes

* linting

* Avoid problem with temp file getting deleted

* Avoid problem with temp file getting deleted

* minor cleanup

* Some more minor fixes

* Fix test

* Fix another test with questionable assumptions

* Dont wait on tmpfile being destroyed

* More regex checks

* Add more postcommits, clean up some error messages

* Clean up a few tests

* simplify test

* Clean up test errors

Author: damccorm

.github/trigger_files/beam_PostCommit_Python.json

@@ -1,5 +1,5 @@
 {
   "comment": "Modify this file in a trivial way to cause this test suite to run.",
-  "modification": 1
+  "modification": 2
 }
 

.github/trigger_files/beam_PostCommit_Python_Dependency.json

@@ -1,4 +1,4 @@
 {
     "comment": "Modify this file in a trivial way to cause this test suite to run",
-    "modification": 1
+    "modification": 2
   }

.github/trigger_files/beam_PostCommit_Python_Xlang_Gcp_Direct.json

@@ -1,4 +1,4 @@
 {
   "comment": "Modify this file in a trivial way to cause this test suite to run",
-  "modification": 8
+  "modification": 9
 }

.github/trigger_files/beam_PostCommit_Python_Xlang_IO_Direct.json

@@ -1,4 +1,4 @@
 {
   "comment": "Modify this file in a trivial way to cause this test suite to run",
-  "modification": 5
+  "modification": 6
 }

.github/workflows/beam_PostCommit_Python_Dependency.yml

@@ -60,7 +60,7 @@ jobs:
         job_name: ['beam_PostCommit_Python_Dependency']
         job_phrase: ['Run Python PostCommit Dependency']
         python_version: ['3.9','3.12']
-    timeout-minutes: 120
+    timeout-minutes: 180
     if: |
       github.event_name == 'workflow_dispatch' ||
       github.event_name == 'pull_request_target' ||

.github/workflows/beam_PostCommit_Python_Xlang_Gcp_Direct.yml

@@ -57,7 +57,7 @@ jobs:
       (github.event_name == 'schedule' && github.repository == 'apache/beam') ||
       github.event.comment.body == 'Run Python_Xlang_Gcp_Direct PostCommit'
     runs-on: [self-hosted, ubuntu-20.04, highmem]
-    timeout-minutes: 100
+    timeout-minutes: 160
     name: ${{ matrix.job_name }} (${{ matrix.job_phrase }})
     strategy:
       matrix:
@@ -98,4 +98,4 @@ jobs:
           commit: '${{ env.prsha || env.GITHUB_SHA }}'
           comment_mode: ${{ github.event_name == 'issue_comment'  && 'always' || 'off' }}
           files: '**/pytest*.xml'
-          large_files: true
+          large_files: true

CHANGES.md

@@ -65,6 +65,7 @@
 
 * New highly anticipated feature X added to Python SDK ([#X](https://github.com/apache/beam/issues/X)).
 * New highly anticipated feature Y added to Java SDK ([#Y](https://github.com/apache/beam/issues/Y)).
+* [Python] Prism runner now enabled by default for most Python pipelines using the direct runner ([#34612](https://github.com/apache/beam/pull/34612)). This may break some tests, see https://github.com/apache/beam/pull/34612 for details on how to handle issues.
 
 ## I/Os
 
@@ -85,6 +86,7 @@
 
 ## Breaking Changes
 
+* [Python] Prism runner now enabled by default for most Python pipelines using the direct runner ([#34612](https://github.com/apache/beam/pull/34612)). This may break some tests, see https://github.com/apache/beam/pull/34612 for details on how to handle issues.
 * X behavior was changed ([#X](https://github.com/apache/beam/issues/X)).
 * Go: The pubsubio.Read transform now accepts ReadOptions as a value type instead of a pointer, and requires exactly one of Topic or Subscription to be set (they are mutually exclusive). Additionally, the ReadOptions struct now includes a Topic field for specifying the topic directly, replacing the previous topic parameter in the Read function signature ([#35369])(https://github.com/apache/beam/pull/35369).
 * SQL: The `ParquetTable` external table provider has changed its handling of the `LOCATION` property. To read from a directory, the path must now end with a trailing slash (e.g., `LOCATION '/path/to/data/'`). Previously, a trailing slash was not required. This change was made to enable support for glob patterns and single-file paths ([#35582])(https://github.com/apache/beam/pull/35582).

sdks/python/apache_beam/coders/coders_test.py

@@ -267,7 +267,7 @@ def test_numpy_int(self):
     # this type is not supported as the key
     import numpy as np
 
-    with self.assertRaises(TypeError):
+    with self.assertRaisesRegex(Exception, "Unable to deterministically"):
       with TestPipeline() as p:
         indata = p | "Create" >> beam.Create([(a, int(a))
                                               for a in np.arange(3)])

sdks/python/apache_beam/examples/snippets/snippets_test.py

@@ -307,8 +307,8 @@ def test_bad_types(self):
     # When running this pipeline, you'd get a runtime error,
     # possibly on a remote machine, possibly very late.
 
-    with self.assertRaises(TypeError):
-      p.run()
+    with self.assertRaisesRegex(Exception, "not all arguments converted"):
+      p.run().wait_until_finish()
 
     # To catch this early, we can assert what types we expect.
     with self.assertRaises(typehints.TypeCheckError):
@@ -372,8 +372,8 @@ def process(self, element):
     # When running this pipeline, you'd get a runtime error,
     # possibly on a remote machine, possibly very late.
 
-    with self.assertRaises(TypeError):
-      p.run()
+    with self.assertRaisesRegex(Exception, "not all arguments converted"):
+      p.run().wait_until_finish()
 
     # To catch this early, we can annotate process() with the expected types.
     # Beam will then use these as type hints and perform type checking before
@@ -439,12 +439,13 @@ def test_runtime_checks_off(self):
 
   def test_runtime_checks_on(self):
     # pylint: disable=expression-not-assigned
-    with self.assertRaises(typehints.TypeCheckError):
+    with self.assertRaisesRegex(Exception, "According to type-hint"):
       # [START type_hints_runtime_on]
       p = TestPipeline(options=PipelineOptions(runtime_type_check=True))
       p | beam.Create(['a']) | beam.Map(lambda x: 3).with_output_types(str)
-      p.run()
+      result = p.run()
       # [END type_hints_runtime_on]
+      result.wait_until_finish()
 
   def test_deterministic_key(self):
     with TestPipeline() as p:

sdks/python/apache_beam/examples/snippets/transforms/elementwise/enrichment_test.py

@@ -60,28 +60,52 @@ def validate_enrichment_with_vertex_ai_legacy():
   return expected
 
 
+def std_out_to_dict(stdout_lines, row_key):
+  output_dict = {}
+  for stdout_line in stdout_lines:
+    # parse the stdout in a dictionary format so that it can be
+    # evaluated/compared as one. This allows us to compare without
+    # considering the order of the stdout or the order that the fields of the
+    # row are arranged in.
+    fmtd = '{\"' + stdout_line[4:-1].replace('=', '\": ').replace(
+        ', ', ', \"').replace('\"\'', '\'') + "}"
+    stdout_dict = eval(fmtd)  # pylint: disable=eval-used
+    output_dict[stdout_dict[row_key]] = stdout_dict
+  return output_dict
+
+
 @mock.patch('sys.stdout', new_callable=StringIO)
 class EnrichmentTest(unittest.TestCase):
   def test_enrichment_with_bigtable(self, mock_stdout):
     enrichment_with_bigtable()
     output = mock_stdout.getvalue().splitlines()
     expected = validate_enrichment_with_bigtable()
-    self.assertEqual(output, expected)
+
+    self.assertEqual(len(output), len(expected))
+    self.assertEqual(
+        std_out_to_dict(output, 'sale_id'),
+        std_out_to_dict(expected, 'sale_id'))
 
   def test_enrichment_with_vertex_ai(self, mock_stdout):
     enrichment_with_vertex_ai()
     output = mock_stdout.getvalue().splitlines()
     expected = validate_enrichment_with_vertex_ai()
 
-    for i in range(len(expected)):
-      self.assertEqual(set(output[i].split(',')), set(expected[i].split(',')))
+    self.assertEqual(len(output), len(expected))
+    self.assertEqual(
+        std_out_to_dict(output, 'user_id'),
+        std_out_to_dict(expected, 'user_id'))
 
   def test_enrichment_with_vertex_ai_legacy(self, mock_stdout):
     enrichment_with_vertex_ai_legacy()
     output = mock_stdout.getvalue().splitlines()
     expected = validate_enrichment_with_vertex_ai_legacy()
     self.maxDiff = None
-    self.assertEqual(output, expected)
+
+    self.assertEqual(len(output), len(expected))
+    self.assertEqual(
+        std_out_to_dict(output, 'entity_id'),
+        std_out_to_dict(expected, 'entity_id'))
 
 
 if __name__ == '__main__':