[BEAM-12030] DataFrame read_* raise FileNotFound for non-existent input

sdks/python/apache_beam/dataframe/io.py

@@ -187,12 +187,14 @@ def __init__(
     self.splitter = splitter
 
   def expand(self, root):
-    # TODO(robertwb): Handle streaming (with explicit schema).
     paths_pcoll = root | beam.Create([self.path])
-    first = io.filesystems.FileSystems.match([self.path],
-                                             limits=[1
-                                                     ])[0].metadata_list[0].path
-    with io.filesystems.FileSystems.open(first) as handle:
+    match = io.filesystems.FileSystems.match([self.path], limits=[1])[0]
+    if not match.metadata_list:
+      # TODO(BEAM-12031): This should be allowed for streaming pipelines if
+      # user provides an explicit schema.
+      raise FileNotFoundError(f"Found no files that match {self.path!r}")
+    first_path = match.metadata_list[0].path
+    with io.filesystems.FileSystems.open(first_path) as handle:
       if not self.binary:
         handle = TextIOWrapper(handle)
       if self.incremental:

sdks/python/apache_beam/dataframe/io_test.py

@@ -299,6 +299,11 @@ def read_truncated_csv(start, stop):
         ])
         assert_frame_equal(expected, split_at_header)
 
+  def test_file_not_found(self):
+    with self.assertRaisesRegex(FileNotFoundError, r'/tmp/fake_dir/\*\*'):
+      with beam.Pipeline() as p:
+        p | io.read_csv('/tmp/fake_dir/**')
+
 
 if __name__ == '__main__':
   unittest.main()