fix(file-based): fire the duplicate-file guard in file-transfer mode

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)
         )
 

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:
         """

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

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):