Repository: deepset-ai/haystack
Pull request: feat: Add raise_on_failure to BaseConverter
### Related Issues

- fixes [#6665](https://github.com/deepset-ai/haystack/issues/6665)

### Proposed Changes:

Followed the potential solution (Thanks @anakin87 !) to add a `raise_on_failure` to the `run` method in the BaseConverter class. 

### How did you test it?

<!-- unit tests, integration tests, manual verification, instructions for manual tests -->
Added unit tests with the CSV converter.

### Notes for the reviewer

<!-- E.g. point out section where the reviewer  -->

### Checklist

- I have read the [contributors guidelines](https://github.com/deepset-ai/haystack/blob/main/CONTRIBUTING.md) and the [code of conduct](https://github.com/deepset-ai/haystack/blob/main/code_of_conduct.txt)
- I have updated the related issue with new insights and changes
- I added unit tests and updated the docstrings
- I've used one of the [conventional commit types](https://www.conventionalcommits.org/en/v1.0.0/) for my PR title: `fix:`, `feat:`, `build:`, `chore:`, `ci:`, `docs:`, `style:`, `refactor:`, `perf:`, `test:`.
- I documented my code
- I ran [pre-commit hooks](https://github.com/deepset-ai/haystack/blob/main/CONTRIBUTING.md#installation) and fixed any issue


Problem statement:
BaseConverter more robust and reliable when processing thousands of files.
**Is your feature request related to a problem? Please describe.**
I'm always frustrated when a conversion suddenly breaks the whole pipeline, especially when processing thousands of files. Imagine importing like 2500 PDF files into an object store via Tika- or PDF-converter. When there is one corrupt PDF or something else goes wrong, an exception is raised breaking the whole process. Of course, typically this happens after one hour has passed and 75% already have been processed. 

**Describe the solution you'd like**
Don't break the whole pipeline or conversion process as soon as there is one exception occuring for whatever reason.


Version: 2.x

Hints from issue discussion:
### Potential solution

Assuming we are talking of 1.x (the `BaseConverter` is only available in 1.x), this does not seem difficult to achieve.
We can add a `raise_on_failure` parameter in the `run method` of the `BaseConverter`.

- if True (default), raises an exception if the conversion of a single file fails (current behavior)
- if False, skips the file without failing

Since the `run` method is present only in the `BaseConverter`, it would be necessary to modify only the latter and not the specific converters.

I will put the tag "Contributions wanted" on this issue. @dannybusch feel free to open a PR...

@anakin87 drafted a PR for this issue. Which branch should I rebase it to? (got quite a big conflict at the moment).

Commit message: Merge branch 'raise_on_failure-base-converter' of github.com:isaac-chung/haystack into raise_on_failure-base-converter

Patch to review:
diff --git a/haystack/nodes/file_converter/base.py b/haystack/nodes/file_converter/base.py
index d5fbb3fc58..9f8109a756 100644
--- a/haystack/nodes/file_converter/base.py
+++ b/haystack/nodes/file_converter/base.py
@@ -158,6 +158,7 @@ def run(  # type: ignore
         valid_languages: Optional[List[str]] = None,
         encoding: Optional[str] = "UTF-8",
         id_hash_keys: Optional[List[str]] = None,
+        raise_on_failure: Optional[bool] = True,
     ):
         """
         Extract text from a file.
@@ -188,6 +189,7 @@ def run(  # type: ignore
             attributes. If you want to ensure you don't have duplicate documents in your DocumentStore but texts are
             not unique, you can modify the metadata and pass e.g. `"meta"` to this field (e.g. [`"content"`, `"meta"`]).
             In this case the id will be generated by using the content and the defined metadata.
+        :param raise_on_failure: If true, raises an exception if the conversion of a single file fails. If False, skips the file without failing.
         """
         if known_ligatures is None:
             known_ligatures = KNOWN_LIGATURES
@@ -199,17 +201,24 @@ def run(  # type: ignore
             meta = [meta] * len(file_paths)
 
         documents: list = []
+        failed_paths: list = []
         for file_path, file_meta in tqdm(
             zip(file_paths, meta), total=len(file_paths), disable=not self.progress_bar, desc="Converting files"
         ):
-            documents += self.convert(
-                file_path=file_path,
-                meta=file_meta,
-                remove_numeric_tables=remove_numeric_tables,
-                valid_languages=valid_languages,
-                encoding=encoding,
-                id_hash_keys=id_hash_keys,
-            )
+            try:
+                documents += self.convert(
+                    file_path=file_path,
+                    meta=file_meta,
+                    remove_numeric_tables=remove_numeric_tables,
+                    valid_languages=valid_languages,
+                    encoding=encoding,
+                    id_hash_keys=id_hash_keys,
+                )
+            except Exception as e:
+                if raise_on_failure:
+                    raise e
+                failed_paths.append(file_path)
+                continue
 
         # Cleanup ligatures
         for document in documents:
@@ -217,6 +226,9 @@ def run(  # type: ignore
                 if document.content is not None:
                     document.content = document.content.replace(ligature, letters)
 
+        if len(failed_paths):
+            logger.warning("Failed paths: %s", ",".join(failed_paths))
+
         result = {"documents": documents}
         return result, "output_1"
 
diff --git a/releasenotes/notes/add-raise-on-failure-to-base-converter-8c5e9b3dd51c0e0c.yaml b/releasenotes/notes/add-raise-on-failure-to-base-converter-8c5e9b3dd51c0e0c.yaml
new file mode 100644
index 0000000000..05e4a959d7
--- /dev/null
+++ b/releasenotes/notes/add-raise-on-failure-to-base-converter-8c5e9b3dd51c0e0c.yaml
@@ -0,0 +1,4 @@
+---
+enhancements:
+  - |
+    Add `raise_on_failure` flag to BaseConverter class so that big processes can optionally continue without breaking from exceptions.
diff --git a/test/nodes/test_file_converter.py b/test/nodes/test_file_converter.py
index 9daee4b587..e9925281c1 100644
--- a/test/nodes/test_file_converter.py
+++ b/test/nodes/test_file_converter.py
@@ -422,6 +422,34 @@ def test_csv_to_document_with_wrong_qa_headers(tmp_path):
         node.run(file_paths=csv_path)
 
 
+@pytest.mark.unit
+def test_csv_to_document_with_wrong_qa_headers_raise_on_failure_true(tmp_path):
+    node = CsvTextConverter()
+    csv_path = tmp_path / "csv_qa_with_wrong_headers.csv"
+    rows = [
+        ["wrong", "headers"],
+        ["What is Haystack ?", "Haystack is an NLP Framework to use transformers in your Applications."],
+    ]
+    write_as_csv(rows, csv_path)
+
+    with pytest.raises(ValueError, match="The CSV must contain two columns named 'question' and 'answer'"):
+        node.run(file_paths=csv_path, raise_on_failure=True)
+
+
+@pytest.mark.unit
+def test_csv_to_document_with_wrong_qa_headers_raise_on_failure_false(tmp_path):
+    node = CsvTextConverter()
+    csv_path = tmp_path / "csv_qa_with_wrong_headers.csv"
+    rows = [
+        ["wrong", "headers"],
+        ["What is Haystack ?", "Haystack is an NLP Framework to use transformers in your Applications."],
+    ]
+    write_as_csv(rows, csv_path)
+
+    result, _ = node.run(file_paths=csv_path, raise_on_failure=False)
+    assert len(result["documents"]) == 0
+
+
 @pytest.mark.unit
 def test_csv_to_document_with_one_wrong_qa_headers(tmp_path):
     node = CsvTextConverter()


Code review comments:
diff_hunk: @@ -199,24 +201,34 @@ def run(  # type: ignore
             meta = [meta] * len(file_paths)
 
         documents: list = []
+        failed_paths: list = []
         for file_path, file_meta in tqdm(
             zip(file_paths, meta), total=len(file_paths), disable=not self.progress_bar, desc="Converting files"
         ):
-            documents += self.convert(
-                file_path=file_path,
-                meta=file_meta,
-                remove_numeric_tables=remove_numeric_tables,
-                valid_languages=valid_languages,
-                encoding=encoding,
-                id_hash_keys=id_hash_keys,
-            )
+            try:
+                documents += self.convert(
+                    file_path=file_path,
+                    meta=file_meta,
+                    remove_numeric_tables=remove_numeric_tables,
+                    valid_languages=valid_languages,
+                    encoding=encoding,
+                    id_hash_keys=id_hash_keys,
+                )
+            except Exception as e:
+                if raise_on_failure:
+                    raise e
+                failed_paths.append(file_path)
+                continue
 
         # Cleanup ligatures
         for document in documents:
             for ligature, letters in known_ligatures.items():
                 if document.content is not None:
                     document.content = document.content.replace(ligature, letters)
 
+        if len(failed_paths):
+            logger.warning("Failed paths: %s", ",".join(failed_paths))
path: haystack/nodes/file_converter/base.py
text: @user1:
```suggestion
        if failed_paths:
            logger.warning("Conversion of the following file paths failed: %s", ",".join(failed_paths))
```