Fix WriteToFiles not creating nested directories for file_naming paths

[Akshaay1]

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