Prism windowed value coder (#34830)

* Support windowed value coder in prism.

* Add a new test case that previously failed to run in prism.

* Fix lints and trigger some validatesrunner tests.

* Disable the newly added test for samza runner.

Author: shunping

.github/trigger_files/beam_PostCommit_Python_ValidatesRunner_Flink.json

@@ -1,3 +1,4 @@
 {
-  "https://github.com/apache/beam/pull/32648": "testing addition of Flink 1.19 support"
+  "https://github.com/apache/beam/pull/32648": "testing addition of Flink 1.19 support",
+  "https://github.com/apache/beam/pull/34830": "testing"
 }

.github/trigger_files/beam_PostCommit_Python_ValidatesRunner_Samza.json

@@ -0,0 +1,3 @@
+{
+  "https://github.com/apache/beam/pull/34830": "testing"
+}

.github/trigger_files/beam_PostCommit_Python_ValidatesRunner_Spark.json

@@ -0,0 +1,3 @@
+{
+  "https://github.com/apache/beam/pull/34830": "testing"
+}

sdks/go/pkg/beam/runners/prism/internal/coders.go

@@ -331,6 +331,17 @@ func pullDecoderNoAlloc(c *pipepb.Coder, coders map[string]*pipepb.Coder) func(i
 			kd(r)
 			vd(r)
 		}
+	case urns.CoderWindowedValue:
+		ccids := c.GetComponentCoderIds()
+		if len(ccids) != 2 {
+			panic(fmt.Sprintf("WindowedValue coder with more than 2 components: %s", prototext.Format(c)))
+		}
+		ed := pullDecoderNoAlloc(coders[ccids[0]], coders)
+		wd := pullDecoderNoAlloc(coders[ccids[1]], coders)
+		return func(r io.Reader) {
+			ed(r)
+			wd(r)
+		}
 	case urns.CoderRow:
 		panic(fmt.Sprintf("Runner forgot to LP this Row Coder. %v", prototext.Format(c)))
 	default:

sdks/python/apache_beam/runners/portability/fn_api_runner/fn_runner_test.py

@@ -1071,6 +1071,15 @@ def test_reshuffle(self):
       assert_that(
           p | beam.Create([1, 2, 3]) | beam.Reshuffle(), equal_to([1, 2, 3]))
 
+  def test_reshuffle_after_custom_window(self):
+    with self.create_pipeline() as p:
+      assert_that(
+          p | beam.Create([12, 2, 1])
+          | beam.Map(lambda t: window.TimestampedValue(t, t))
+          | beam.WindowInto(beam.transforms.window.FixedWindows(2))
+          | beam.Reshuffle(),
+          equal_to([12, 2, 1]))
+
   def test_flatten(self, with_transcoding=True):
     with self.create_pipeline() as p:
       if with_transcoding:

sdks/python/apache_beam/runners/portability/samza_runner_test.py

@@ -186,6 +186,9 @@ def test_custom_merging_window(self):
   def test_custom_window_type(self):
     raise unittest.SkipTest("https://github.com/apache/beam/issues/21049")
 
+  def test_reshuffle_after_custom_window(self):
+    raise unittest.SkipTest("https://github.com/apache/beam/issues/34831")
+
 
 if __name__ == '__main__':
   # Run the tests.