fail windowInto when there are no inputs to it

sdks/python/apache_beam/pipeline.py

@@ -1353,18 +1353,26 @@ def transform_to_runner_api(
       environment_id = context.get_environment_id_for_resource_hints(
           self.resource_hints)
 
+    inputs = {
+        tag: context.pcollections.get_id(pc)
+        for tag,
+        pc in sorted(self.named_inputs().items())
+    }
+
+    if 'WindowInto(WindowIntoFn)' in self.full_label and not inputs:
+      raise ValueError(
+          "No input PCollection for WindowInto. "
+          "Please make sure that beam.WindowInto follows "
+          "a PCollection in the pipeline.")
+
     return beam_runner_api_pb2.PTransform(
         unique_name=self.full_label,
         spec=transform_spec,
         subtransforms=[
             context.transforms.get_id(part, label=part.full_label)
             for part in self.parts
         ],
-        inputs={
-            tag: context.pcollections.get_id(pc)
-            for tag,
-            pc in sorted(self.named_inputs().items())
-        },
+        inputs=inputs,
         outputs={
             tag: context.pcollections.get_id(out)
             for tag,

sdks/python/apache_beam/pipeline_test.py

@@ -1337,6 +1337,12 @@ def file_artifact(path, hash, staged_name):
 
     self.assertEqual(len(proto.components.environments), 6)
 
+  def test_window_into_fail_pipeline(self):
+    with self.assertRaises(ValueError):
+      p = beam.Pipeline()
+      _ = p | beam.WindowInto(beam.window.FixedWindows(2 * 24 * 60 * 60))
+      _ = Pipeline.to_runner_api(p, use_fake_coders=True)
+
 
 if __name__ == '__main__':
   unittest.main()