From 7835d7bcd12cb59ffd701dceb4a86c6984e36dcd Mon Sep 17 00:00:00 2001
From: Akshaay <89546672+Akshaay1@users.noreply.github.com>
Date: Mon, 12 Oct 2026 06:49:42 +0000
Subject: [PATCH] Fix WriteToFiles not creating nested directories for
 file_naming paths

WriteToFiles only created the base output path before renaming temp
files into their final destinations. When file_naming returned a path
with a subdirectory component, the rename failed because the target
directory never existed, silently dropping the written files.

Fixes #21313
---
 CHANGES.md                                |  1 +
 sdks/python/apache_beam/io/fileio.py      | 28 ++++++++++++++---------
 sdks/python/apache_beam/io/fileio_test.py | 24 +++++++++++++++++++
 3 files changed, 42 insertions(+), 11 deletions(-)

diff --git a/CHANGES.md b/CHANGES.md
index cb5eced42f7c..2a8b06af2e83 100644
--- a/CHANGES.md
+++ b/CHANGES.md
@@ -89,6 +89,7 @@
 
 ## Bugfixes
 
+* (Python) Fixed `WriteToFiles` failing to write files when `file_naming` places them in a subdirectory of `path`, since only `path` itself was created before the move ([#21313](https://github.com/apache/beam/issues/21313)).
 * (Go) Fixed a data race on the Prism runner's artifact cache map in JobServices ([#32656](https://github.com/apache/beam/issues/32656)).
 * (Go) Fixed the harness leaking Data/State gRPC streams after the worker stops, and a deadlock when Send returns EOF ([#40260](https://github.com/apache/beam/issues/40260)).
 * (Go) Fixed a deadlock recreating a data channel while holding the channel lock ([#40414](https://github.com/apache/beam/issues/40414)).
diff --git a/sdks/python/apache_beam/io/fileio.py b/sdks/python/apache_beam/io/fileio.py
index fcce83fa59ec..0197b746bb02 100644
--- a/sdks/python/apache_beam/io/fileio.py
+++ b/sdks/python/apache_beam/io/fileio.py
@@ -873,7 +873,10 @@ def process(self, element, w=beam.DoFn.WindowParam):
               destination))
 
     move_from = [f.file_name for f in temp_file_results]
-    move_to = [f.file_name for f in final_file_results]
+    move_to = [
+        filesystems.FileSystems.join(self.path.get(), f.file_name)
+        for f in final_file_results
+    ]
 
     _LOGGER.info(
         'Moving %d temporary files to dir: %s as %s',
@@ -881,18 +884,21 @@ def process(self, element, w=beam.DoFn.WindowParam):
         self.path.get(),
         move_to)
 
-    try:
-      filesystems.FileSystems.mkdirs(self.path.get())
-    except IOError as e:
-      cause = repr(e)
-      if 'FileExistsError' not in cause:
-        # Usually harmless. Especially if see FileExistsError so no need to log
-        _LOGGER.debug('Fail to create dir for final destination: %s', cause)
+    # file_naming_fn may nest the final file name under a subdirectory of
+    # self.path, so create the parent directory of each final destination
+    # rather than just the base path.
+    for final_dir in {filesystems.FileSystems.split(move_to_path)[0]
+                      for move_to_path in move_to}:
+      try:
+        filesystems.FileSystems.mkdirs(final_dir)
+      except IOError as e:
+        cause = repr(e)
+        if 'FileExistsError' not in cause:
+          # Usually harmless. Especially if see FileExistsError so no need to log
+          _LOGGER.debug('Fail to create dir for final destination: %s', cause)
 
     try:
-      filesystems.FileSystems.rename(
-          move_from,
-          [filesystems.FileSystems.join(self.path.get(), f) for f in move_to])
+      filesystems.FileSystems.rename(move_from, move_to)
     except BeamIOError:
       # This error is not serious, because it may happen on a retry of the
       # bundle. We simply log it.
diff --git a/sdks/python/apache_beam/io/fileio_test.py b/sdks/python/apache_beam/io/fileio_test.py
index 1c650653804c..7fd244cc0114 100644
--- a/sdks/python/apache_beam/io/fileio_test.py
+++ b/sdks/python/apache_beam/io/fileio_test.py
@@ -879,6 +879,30 @@ def test_write_to_dynamic_destination(self):
             equal_to([('odd', list(range(1, 100, 2))),
                       ('even', list(range(0, 100, 2)))]))
 
+  def test_write_to_nested_directory_destination(self):
+    dir = self._new_tempdir()
+
+    def nested_file_naming(window, pane, shard_index, total_shards, *args):
+      return FileSystems.join('subdir', 'test_file_%s' % shard_index)
+
+    with TestPipeline() as p:
+      _ = (
+          p
+          | "Create" >> beam.Create(range(10))
+          | beam.Map(lambda x: str(x))
+          | fileio.WriteToFiles(
+              path=dir, sink=fileio.TextSink, file_naming=nested_file_naming))
+
+    with TestPipeline() as p:
+      result = (
+          p
+          | fileio.MatchFiles(FileSystems.join(dir, 'subdir', '*'))
+          | fileio.ReadMatches()
+          | beam.FlatMap(lambda f: f.read_utf8().strip().split('\n'))
+          | beam.Map(int))
+
+      assert_that(result, equal_to(list(range(10))))
+
   def test_write_to_different_file_types_some_spilling(self):
 
     dir = self._new_tempdir()