From 0388bfd79efbbcbc5cb59b8472c764c5cc75d83e Mon Sep 17 00:00:00 2001 From: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com> Date: Sun, 26 Jul 2026 11:50:55 +0000 Subject: [PATCH] fix(file-based): namespace file transfer staging paths per file Two source files that resolve to the same relative path shared one staging file, so the second download overwrote the first and the destination failed with FileNotFoundException after consuming and deleting the first copy. Co-Authored-By: bot_apk --- .../file_based/file_based_stream_reader.py | 8 ++- .../test_file_based_stream_reader.py | 66 +++++++++++++++++-- 2 files changed, 66 insertions(+), 8 deletions(-) diff --git a/airbyte_cdk/sources/file_based/file_based_stream_reader.py b/airbyte_cdk/sources/file_based/file_based_stream_reader.py index f39d2860c0..8a06ac7142 100644 --- a/airbyte_cdk/sources/file_based/file_based_stream_reader.py +++ b/airbyte_cdk/sources/file_based/file_based_stream_reader.py @@ -4,6 +4,7 @@ import logging import time +import uuid from abc import ABC, abstractmethod from datetime import datetime, timezone from enum import Enum @@ -254,6 +255,11 @@ def _get_file_transfer_paths( - LOCAL_FILE_PATH: The absolute path to the file. - FILE_NAME: The name of the referenced file. - FILE_FOLDER: The folder of the referenced file. + + The local path is namespaced under a unique staging subdirectory so that two source files + resolving to the same relative path never share a staging file. Sharing a staging file makes + one download overwrite the other and makes the destination fail once it has consumed and + deleted the first copy. """ preserve_directory_structure = self.preserve_directory_structure() @@ -264,7 +270,7 @@ def _get_file_transfer_paths( file_relative_path = source_file_relative_path.lstrip("/") else: file_relative_path = file_name - local_file_path = path.join(staging_directory, file_relative_path) + local_file_path = path.join(staging_directory, uuid.uuid4().hex, file_relative_path) # Ensure the local directory exists makedirs(path.dirname(local_file_path), exist_ok=True) diff --git a/unit_tests/sources/file_based/test_file_based_stream_reader.py b/unit_tests/sources/file_based/test_file_based_stream_reader.py index a8b8e7752f..fc473545cb 100644 --- a/unit_tests/sources/file_based/test_file_based_stream_reader.py +++ b/unit_tests/sources/file_based/test_file_based_stream_reader.py @@ -433,7 +433,7 @@ def test_globs_and_prefixes_from_globs( @pytest.mark.parametrize( - "config, source_file_path, expected_file_relative_path, expected_local_file_path", + "config, source_file_path, expected_file_relative_path", [ pytest.param( { @@ -445,7 +445,6 @@ def test_globs_and_prefixes_from_globs( }, "mirror_paths_testing/not_duplicates/data/jan/monthly-kickoff-202402.mpeg", "mirror_paths_testing/not_duplicates/data/jan/monthly-kickoff-202402.mpeg", - f"{files_directory}/mirror_paths_testing/not_duplicates/data/jan/monthly-kickoff-202402.mpeg", id="preserve_directories_present_and_true", ), pytest.param( @@ -458,21 +457,18 @@ def test_globs_and_prefixes_from_globs( }, "mirror_paths_testing/not_duplicates/data/jan/monthly-kickoff-202402.mpeg", "monthly-kickoff-202402.mpeg", - f"{files_directory}/monthly-kickoff-202402.mpeg", id="preserve_directories_present_and_false", ), pytest.param( {"streams": [], "delivery_method": {"delivery_type": "use_file_transfer"}}, "mirror_paths_testing/not_duplicates/data/jan/monthly-kickoff-202402.mpeg", "mirror_paths_testing/not_duplicates/data/jan/monthly-kickoff-202402.mpeg", - f"{files_directory}/mirror_paths_testing/not_duplicates/data/jan/monthly-kickoff-202402.mpeg", id="preserve_directories_not_present_defaults_true", ), pytest.param( {"streams": []}, "mirror_paths_testing/not_duplicates/data/jan/monthly-kickoff-202402.mpeg", "mirror_paths_testing/not_duplicates/data/jan/monthly-kickoff-202402.mpeg", - f"{files_directory}/mirror_paths_testing/not_duplicates/data/jan/monthly-kickoff-202402.mpeg", id="file_transfer_flag_not_present_defaults_true", ), ], @@ -481,7 +477,6 @@ def test_preserve_sub_directories_scenarios( config: Mapping[str, Any], source_file_path: str, expected_file_relative_path: str, - expected_local_file_path: str, ) -> None: """ Test scenarios when preserve_directory_structure is True or False, the flag indicates whether we need to @@ -496,11 +491,68 @@ def test_preserve_sub_directories_scenarios( assert ( file_paths[AbstractFileBasedStreamReader.FILE_RELATIVE_PATH] == expected_file_relative_path ) - assert file_paths[AbstractFileBasedStreamReader.LOCAL_FILE_PATH] == expected_local_file_path + local_file_path = file_paths[AbstractFileBasedStreamReader.LOCAL_FILE_PATH] + staging_subdirectory = path.relpath(local_file_path, files_directory).split(path.sep)[0] + assert local_file_path == path.join( + files_directory, staging_subdirectory, expected_file_relative_path + ) assert file_paths[AbstractFileBasedStreamReader.FILE_NAME] == path.basename(source_file_path) assert file_paths[AbstractFileBasedStreamReader.FILE_FOLDER] == path.dirname(source_file_path) +@pytest.mark.parametrize( + "preserve_directory_structure, first_source_file_path, second_source_file_path", + [ + pytest.param( + True, + "folder_a/Untitled document.docx", + "folder_a/Untitled document.docx", + id="same_relative_path_from_distinct_source_files", + ), + pytest.param( + False, + "folder_a/Untitled document.docx", + "folder_b/Untitled document.docx", + id="same_file_name_in_different_folders_without_preserved_directories", + ), + ], +) +def test_staging_paths_are_unique_per_file( + preserve_directory_structure: bool, + first_source_file_path: str, + second_source_file_path: str, +) -> None: + """ + Two source files that resolve to the same relative path must not share a staging file: the second + download would overwrite the first, and the destination fails with a FileNotFoundException once it + has consumed and deleted the first copy. + """ + reader = TestStreamReader() + reader.config = TestSpec( + streams=[], + delivery_method={ + "delivery_type": "use_file_transfer", + "preserve_directory_structure": preserve_directory_structure, + }, + ) + + first_paths = reader._get_file_transfer_paths( + first_source_file_path, staging_directory=f"{files_directory}/" + ) + second_paths = reader._get_file_transfer_paths( + second_source_file_path, staging_directory=f"{files_directory}/" + ) + + assert ( + first_paths[AbstractFileBasedStreamReader.FILE_RELATIVE_PATH] + == second_paths[AbstractFileBasedStreamReader.FILE_RELATIVE_PATH] + ) + assert ( + first_paths[AbstractFileBasedStreamReader.LOCAL_FILE_PATH] + != second_paths[AbstractFileBasedStreamReader.LOCAL_FILE_PATH] + ) + + @pytest.mark.parametrize( "start_date_str, expected", [