[PR] Fix WriteToFiles not creating nested directories for file_naming paths [beam]

["Akshaay1 (via GitHub)" <gi...@apache.org>]

Akshaay1 opened a new pull request, #40514:
URL: https://github.com/apache/beam/pull/40514

   ## Problem
   
   `WriteToFiles` fails to write output when the `file_naming` function places the final file under a subdirectory of `path` (e.g. `file_naming=lambda *args: "subdir/file.txt"`). The write silently produces no files instead of raising, because the rename error is caught and only logged at debug level.
   
   ## Root Cause
   
   `_MoveTempFilesIntoFinalDestinationFn.process()` in `sdks/python/apache_beam/io/fileio.py` calls `FileSystems.mkdirs(self.path.get())` before renaming temp files into their final location, but the actual rename targets are `FileSystems.join(self.path.get(), final_file_name)`, where `final_file_name` comes from the user-supplied `file_naming` function and can itself contain a subdirectory. Only the base `path` gets created, not the nested directory the final file actually lives in. On filesystems that require the destination directory to exist before a rename (e.g. the local filesystem, via `os.rename`), this raises an `IOError`/`BeamIOError` that's caught and logged at debug level, so the files are silently never written.
   
   ## Solution
   
   Compute the final destination paths first, then create the parent directory of each unique final destination (via `FileSystems.split`) before attempting the rename, instead of only creating the base path.
   
   ## Testing
   
   - Test added: `test_write_to_nested_directory_destination` in `sdks/python/apache_beam/io/fileio_test.py`, using a `file_naming` function that nests output under a `subdir/` directory and asserting all records are read back.
   - Tests run:
     - Confirmed the new test fails before the fix (`BeamAssertException: ... == [], missing elements [...]`, i.e. zero files were written to the subdirectory).
     - `pytest apache_beam/io/fileio_test.py::WriteFilesTest` — 8 passed, 1 pre-existing skip, after the fix.
     - `pytest apache_beam/io/fileio_test.py` (full file) — 45 passed, 2 pre-existing skips.
     - `yapf --diff` clean on both changed Python files.
   - Result: all pass.
   
   ## Related Issue
   
   Fixes #21313
   


-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: github-unsubscribe@beam.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org