diff --git a/airbyte_cdk/sources/file_based/exceptions.py b/airbyte_cdk/sources/file_based/exceptions.py index 5953850b26..ecfb10a4af 100644 --- a/airbyte_cdk/sources/file_based/exceptions.py +++ b/airbyte_cdk/sources/file_based/exceptions.py @@ -133,15 +133,15 @@ def _format_duplicate_files_error_message(self) -> str: for duplicated_file in self._duplicated_files_names: for duplicated_file_name, file_paths in duplicated_file.items(): file_duplicated_message = ( - f"{len(file_paths)} duplicates found for file name {duplicated_file_name}:\n\n" + f"{len(file_paths)} duplicates found for resolved output path {duplicated_file_name}:\n\n" + "".join(f"\n - {file_paths}") ) duplicated_files_messages.append(file_duplicated_message) error_message = ( - f"ERROR: Duplicate filenames found for stream {self._stream_name}. " - "Duplicate file names are not allowed if the Preserve Sub-Directories in File Paths option is disabled. " - "Please remove or rename the duplicate files before attempting to re-run the sync.\n\n" + f"ERROR: Duplicate files found for stream {self._stream_name}. " + "Multiple source files resolve to the same output path. " + "Rename or remove one of the conflicting source files before attempting to re-run the sync.\n\n" + "\n".join(duplicated_files_messages) ) diff --git a/airbyte_cdk/sources/file_based/remote_file.py b/airbyte_cdk/sources/file_based/remote_file.py index 8d30a53330..65a05e3a7c 100644 --- a/airbyte_cdk/sources/file_based/remote_file.py +++ b/airbyte_cdk/sources/file_based/remote_file.py @@ -17,6 +17,13 @@ class RemoteFile(BaseModel): last_modified: datetime mime_type: Optional[str] = None + @property + def source_file_relative_path(self) -> str: + """ + Returns the relative path of the source file. + """ + return self.uri + @property def file_uri_for_logging(self) -> str: """Returns a user-friendly identifier for logging.""" @@ -47,13 +54,6 @@ def download_to_local_directory(self, local_file_path: str) -> None: """ ... - @property - def source_file_relative_path(self) -> str: - """ - Returns the relative path of the source file. - """ - return self.uri - @property def source_uri(self) -> str: """ diff --git a/airbyte_cdk/sources/file_based/stream/default_file_based_stream.py b/airbyte_cdk/sources/file_based/stream/default_file_based_stream.py index 588c4a18e4..d20ed840bc 100644 --- a/airbyte_cdk/sources/file_based/stream/default_file_based_stream.py +++ b/airbyte_cdk/sources/file_based/stream/default_file_based_stream.py @@ -107,14 +107,16 @@ def primary_key(self) -> PrimaryKeyType: self.config ) - def _duplicated_files_names( - self, slices: List[dict[str, List[RemoteFile]]] - ) -> List[dict[str, List[str]]]: + def _duplicated_files_names(self, files: List[RemoteFile]) -> List[dict[str, List[str]]]: seen_file_names: Dict[str, List[str]] = defaultdict(list) - for file_slice in slices: - for file_found in file_slice[self.FILES_KEY]: - file_name = path.basename(file_found.uri) - seen_file_names[file_name].append(file_found.uri) + for file_found in files: + source_file_relative_path = file_found.source_file_relative_path + file_name = ( + source_file_relative_path.lstrip("/") + if self.preserve_directory_structure + else path.basename(source_file_relative_path) + ) + seen_file_names[file_name].append(file_found.uri) return [ {file_name: paths} for file_name, paths in seen_file_names.items() if len(paths) > 1 ] @@ -128,8 +130,8 @@ def compute_slices(self) -> Iterable[Optional[Mapping[str, Any]]]: {self.FILES_KEY: list(group[1])} for group in itertools.groupby(sorted_files_to_read, lambda f: f.last_modified) ] - if slices and not self.preserve_directory_structure: - duplicated_files_names = self._duplicated_files_names(slices) + if all_files and self.use_file_transfer: + duplicated_files_names = self._duplicated_files_names(all_files) if duplicated_files_names: raise DuplicatedFilesError( stream=self.name, duplicated_files_names=duplicated_files_names diff --git a/unit_tests/sources/file_based/stream/test_default_file_based_stream.py b/unit_tests/sources/file_based/stream/test_default_file_based_stream.py index 54394a36d8..af3c422f32 100644 --- a/unit_tests/sources/file_based/stream/test_default_file_based_stream.py +++ b/unit_tests/sources/file_based/stream/test_default_file_based_stream.py @@ -395,6 +395,64 @@ def test_when_compute_slices(self) -> None: {"files": sorted(all_files, key=lambda f: (f.last_modified, f.uri))} ] + def test_duplicate_uris_raise_when_preserving_directory_structure(self) -> None: + all_files = [ + RemoteFile(uri="same-uri", last_modified=self._NOW), + RemoteFile(uri="same-uri", last_modified=self._NOW), + ] + with ( + mock.patch.object(DefaultFileBasedStream, "list_files", return_value=all_files), + mock.patch.object(self._stream._cursor, "get_files_to_sync", return_value=all_files), + pytest.raises(DuplicatedFilesError), + ): + self._stream.compute_slices() + + def test_same_basename_different_folders_allowed_when_preserving(self) -> None: + all_files = [ + RemoteFile(uri="folder_a/file.csv", last_modified=self._NOW), + RemoteFile(uri="folder_b/file.csv", last_modified=self._NOW), + ] + with ( + mock.patch.object(DefaultFileBasedStream, "list_files", return_value=all_files), + mock.patch.object(self._stream._cursor, "get_files_to_sync", return_value=all_files), + ): + self._stream.compute_slices() + + def test_records_mode_never_raises_on_duplicate_uris(self) -> None: + self._stream.use_file_transfer = False + all_files = [ + RemoteFile(uri="same-uri", last_modified=self._NOW), + RemoteFile(uri="same-uri", last_modified=self._NOW), + ] + with ( + mock.patch.object(DefaultFileBasedStream, "list_files", return_value=all_files), + mock.patch.object(self._stream._cursor, "get_files_to_sync", return_value=all_files), + ): + self._stream.compute_slices() + + def test_guard_compares_source_file_relative_path(self) -> None: + class RemoteFileWithRelativePath(RemoteFile): + relative_path: str + + @property + def source_file_relative_path(self) -> str: + return self.relative_path + + all_files = [ + RemoteFileWithRelativePath( + uri="source-a", relative_path="same/path.csv", last_modified=self._NOW + ), + RemoteFileWithRelativePath( + uri="source-b", relative_path="same/path.csv", last_modified=self._NOW + ), + ] + with ( + mock.patch.object(DefaultFileBasedStream, "list_files", return_value=all_files), + mock.patch.object(self._stream._cursor, "get_files_to_sync", return_value=all_files), + pytest.raises(DuplicatedFilesError), + ): + self._stream.compute_slices() + class DefaultFileBasedStreamFileTransferTestNotMirroringDirectories(unittest.TestCase): _NOW = datetime(2022, 10, 22, tzinfo=timezone.utc) @@ -489,13 +547,40 @@ def test_when_compute_slices_with_duplicates(self) -> None: with ( mock.patch.object(DefaultFileBasedStream, "list_files", return_value=all_files), mock.patch.object(self._stream._cursor, "get_files_to_sync", return_value=all_files), + pytest.raises(DuplicatedFilesError) as exc_info, + ): + self._stream.compute_slices() + assert "Duplicate files found for stream" in str(exc_info.value) + assert "2 duplicates found for resolved output path monthly-kickoff-202402.mpeg" in str( + exc_info.value + ) + assert "2 duplicates found for resolved output path monthly-kickoff-202401.mpeg" in str( + exc_info.value + ) + assert "3 duplicates found for resolved output path monthly-kickoff-202403.mpeg" in str( + exc_info.value + ) + + def test_duplicate_detected_across_incremental_syncs(self) -> None: + all_files = [ + RemoteFile( + uri="folder_a/file.csv", + last_modified=datetime(2025, 1, 9, 11, 27, 20), + ), + RemoteFile( + uri="folder_b/file.csv", + last_modified=datetime(2025, 1, 9, 11, 27, 20), + ), + ] + files_to_read = [all_files[1]] + with ( + mock.patch.object(DefaultFileBasedStream, "list_files", return_value=all_files), + mock.patch.object( + self._stream._cursor, "get_files_to_sync", return_value=files_to_read + ), + pytest.raises(DuplicatedFilesError), ): - with pytest.raises(DuplicatedFilesError) as exc_info: - self._stream.compute_slices() - assert "Duplicate filenames found for stream" in str(exc_info.value) - assert "2 duplicates found for file name monthly-kickoff-202402.mpeg" in str(exc_info.value) - assert "2 duplicates found for file name monthly-kickoff-202401.mpeg" in str(exc_info.value) - assert "3 duplicates found for file name monthly-kickoff-202403.mpeg" in str(exc_info.value) + self._stream.compute_slices() class DefaultFileBasedStreamSchemaTest(unittest.TestCase):