feat(opendataloader_pdf): add optional batch bisection to skip bad PDFs

integrations/opendataloader_pdf/src/haystack_integrations/components/converters/opendataloader_pdf/converter.py

@@ -54,6 +54,7 @@ def __init__(
         convert_kwargs: dict[str, Any] | None = None,
         extract_images: bool = False,
         image_output_dir: str | Path | None = None,
+        skip_failed_pdfs: bool = False,
     ) -> None:
         """
         Initialize the OpenDataLoader converter.
@@ -65,6 +66,8 @@ def __init__(
         :param extract_images: Whether to extract images and return them through the `image_documents` output.
         :param image_output_dir: Persistent root directory for extracted image files. Each `run()` stores its images
             in a unique subdirectory of this directory. Required when `extract_images` is `True`.
+        :param skip_failed_pdfs: Bisect failed batches and skip failing PDFs with a warning. Java and source
+            preparation errors still propagate.
         :raises ValueError: If image extraction is enabled without an output directory.
         """
         conversion_options = convert_kwargs.copy() if convert_kwargs else {}
@@ -83,6 +86,7 @@ def __init__(
         self.convert_kwargs = conversion_options
         self.extract_images = extract_images
         self.image_output_dir = Path(image_output_dir) if image_output_dir is not None else None
+        self.skip_failed_pdfs = skip_failed_pdfs
 
     def to_dict(self) -> dict[str, Any]:
         """
@@ -97,6 +101,7 @@ def to_dict(self) -> dict[str, Any]:
             convert_kwargs=self.convert_kwargs,
             extract_images=self.extract_images,
             image_output_dir=str(self.image_output_dir) if self.image_output_dir is not None else None,
+            skip_failed_pdfs=self.skip_failed_pdfs,
         )
 
     @classmethod
@@ -188,6 +193,63 @@ def _read_output(self, output_dir: Path, pdf_path: Path) -> str:
             raise RuntimeError(message)
         return output_file.read_text(encoding="utf-8")
 
+    def _convert_batch(
+        self,
+        sources: list[tuple[Path, dict[str, Any]]],
+        output_dir: Path,
+        image_dir: Path | None,
+    ) -> list[Document]:
+        """Convert a non-empty batch, bisecting failures into isolated output directories when enabled."""
+        output_dir.mkdir(parents=True, exist_ok=True)
+        conversion_kwargs = {"image_output": "off", **self.convert_kwargs}
+        if image_dir is not None:
+            image_dir.mkdir(parents=True, exist_ok=True)
+            conversion_kwargs.update(image_output="external", image_dir=str(image_dir))
+
+        try:
+            opendataloader_pdf.convert(
+                input_path=[str(pdf_path) for pdf_path, _ in sources],
+                output_dir=str(output_dir),
+                format=self.output_format,
+                **conversion_kwargs,
+            )
+            contents = [self._read_output(output_dir=output_dir, pdf_path=pdf_path) for pdf_path, _ in sources]
+        except (subprocess.CalledProcessError, RuntimeError) as exc:
+            if not self.skip_failed_pdfs:
+                raise
+            # Failed attempts may leave seemingly valid text and images. Never reuse their outputs.
+            shutil.rmtree(output_dir)
+            if image_dir is not None:
+                shutil.rmtree(image_dir)
+            if len(sources) == 1:
+                pdf_path, document_meta = sources[0]
+                logger.warning(
+                    "Skipping PDF {file_path} after conversion failure: {error}",
+                    file_path=document_meta.get("file_path", str(pdf_path)),
+                    error=exc,
+                )
+                return []
+
+            logger.warning(
+                "Batch of {count} PDFs failed; splitting into smaller batches: {error}", count=len(sources), error=exc
+            )
+            midpoint = len(sources) // 2
+            documents = []
+            for name, batch in (("left", sources[:midpoint]), ("right", sources[midpoint:])):
+                documents.extend(
+                    self._convert_batch(
+                        sources=batch,
+                        output_dir=output_dir / name,
+                        image_dir=image_dir / name if image_dir is not None else None,
+                    )
+                )
+            return documents
+
+        return [
+            Document(content=content, meta={**document_meta, "output_format": self.output_format})
+            for (_, document_meta), content in zip(sources, contents, strict=True)
+        ]
+
     @component.output_types(documents=list[Document], image_documents=list[Document])
     def run(
         self,
@@ -202,7 +264,8 @@ def run(
             source. A list must contain one dictionary per source. ByteStream metadata is also preserved.
         :returns:
             Dictionary containing the converted text Documents and image Documents. Each image Document has the
-            persistent extracted image path in its `file_path` metadata field.
+            persistent extracted image path in its `file_path` metadata field. With `skip_failed_pdfs=True`, PDFs
+            that fail individual conversion are omitted from both outputs.
         """
         if not sources:
             return {"documents": [], "image_documents": []}
@@ -226,19 +289,7 @@ def run(
             output_dir.mkdir()
             prepared_sources = self._prepare_sources(sources=sources, metadata=metadata, input_dir=input_dir)
 
-            conversion_kwargs = {"image_output": "off", **self.convert_kwargs}
-            if image_dir is not None:
-                conversion_kwargs.update(image_output="external", image_dir=str(image_dir))
-
-            opendataloader_pdf.convert(
-                input_path=[str(pdf_path) for pdf_path, _ in prepared_sources],
-                output_dir=str(output_dir),
-                format=self.output_format,
-                **conversion_kwargs,
-            )
-            for pdf_path, document_meta in prepared_sources:
-                content = self._read_output(output_dir=output_dir, pdf_path=pdf_path)
-                documents.append(Document(content=content, meta={**document_meta, "output_format": self.output_format}))
+            documents = self._convert_batch(sources=prepared_sources, output_dir=output_dir, image_dir=image_dir)
 
         if image_dir is not None:
             image_documents = [

integrations/opendataloader_pdf/tests/test_converter.py

@@ -1,3 +1,7 @@
+# SPDX-FileCopyrightText: 2026-present deepset GmbH <info@deepset.ai>
+#
+# SPDX-License-Identifier: Apache-2.0
+
 import subprocess
 from concurrent.futures import ThreadPoolExecutor
 from pathlib import Path
@@ -429,13 +433,160 @@ def fail_java_check():
     assert not (tmp_path / "images").exists()
 
 
+def test_converter_bisects_failed_batches_and_cleans_partial_outputs(
+    tmp_path, _mock_opendataloader, monkeypatch, caplog
+):
+    files = [tmp_path / name for name in ("good.pdf", "bad.pdf", "another.pdf", "fourth.pdf", "fifth.pdf")]
+    for pdf in files:
+        pdf.write_bytes(b"%PDF fake pdf")
+    convert = converter_module.opendataloader_pdf.convert
+    calls = []
+
+    def failing_convert(input_path, output_dir, **kwargs):
+        calls.append([Path(path).stem for path in input_path])
+        assert kwargs["format"] == "text"
+        assert kwargs["language"] == "en"
+        convert(input_path, output_dir, **kwargs)
+        if "document_1" in calls[-1]:
+            (Path(kwargs["image_dir"]) / "partial.png").write_bytes(b"partial")
+            message = "corrupt PDF"
+            raise RuntimeError(message)
+
+    monkeypatch.setattr(converter_module.opendataloader_pdf, "convert", failing_convert)
+    converter = OpenDataLoaderConverter(
+        skip_failed_pdfs=True,
+        extract_images=True,
+        image_output_dir=tmp_path / "images",
+        output_format="text",
+        convert_kwargs={"language": "en"},
+    )
+    result = converter.run(sources=files, meta=[{"index": index} for index in range(len(files))])
+
+    assert calls == [
+        ["document_0", "document_1", "document_2", "document_3", "document_4"],
+        ["document_0", "document_1"],
+        ["document_0"],
+        ["document_1"],
+        ["document_2", "document_3", "document_4"],
+    ]
+    assert [document.meta["file_path"] for document in result["documents"]] == [
+        "good.pdf",
+        "another.pdf",
+        "fourth.pdf",
+        "fifth.pdf",
+    ]
+    assert [document.meta["index"] for document in result["documents"]] == [0, 2, 3, 4]
+    assert all(document.meta["output_format"] == "text" for document in result["documents"])
+    image_paths = [Path(document.meta["file_path"]) for document in result["image_documents"]]
+    assert len(image_paths) == 4
+    assert {path.name for path in image_paths} == {f"document_{index}_image_1.png" for index in (0, 2, 3, 4)}
+    assert set((tmp_path / "images").rglob("*.png")) == set(image_paths)
+    assert all(path.read_bytes() == b"fake image" for path in image_paths)
+    assert "Skipping PDF bad.pdf" in caplog.text
+    assert "splitting into smaller batches" in caplog.text
+
+
+def test_converter_default_propagates_batch_failure(tmp_path, _mock_opendataloader, monkeypatch):
+    pdf_file = tmp_path / "bad.pdf"
+    pdf_file.write_bytes(b"%PDF fake pdf")
+    calls = []
+
+    def fail_convert(*, input_path, **_kwargs):
+        calls.append(input_path)
+        message = "corrupt PDF"
+        raise RuntimeError(message)
+
+    monkeypatch.setattr(converter_module.opendataloader_pdf, "convert", fail_convert)
+    with pytest.raises(RuntimeError, match="corrupt PDF"):
+        OpenDataLoaderConverter().run(sources=[pdf_file])
+    assert len(calls) == 1
+
+
+@pytest.mark.parametrize(
+    ("count", "bad_indices", "expected_calls"),
+    [(8, (), 1), (1024, (513,), 21), (5, (1, 3), 9), (4, (0, 1, 2, 3), 7), (1, (0,), 1)],
+)
+def test_converter_bisection_call_count(count, bad_indices, expected_calls, _mock_opendataloader, monkeypatch, caplog):
+    convert = converter_module.opendataloader_pdf.convert
+    calls = []
+    sources = [
+        ByteStream(data=b"%PDF fake pdf", meta={"file_path": f"upload/{index}.pdf", "index": index})
+        for index in range(count)
+    ]
+
+    def failing_convert(input_path, output_dir, **kwargs):
+        calls.append(input_path)
+        if any(Path(path).stem == f"document_{index}" for path in input_path for index in bad_indices):
+            message = "corrupt PDF"
+            raise subprocess.CalledProcessError(returncode=1, cmd=["java"], stderr=message)
+        convert(input_path, output_dir, **kwargs)
+
+    monkeypatch.setattr(converter_module.opendataloader_pdf, "convert", failing_convert)
+    result = OpenDataLoaderConverter(skip_failed_pdfs=True).run(sources=sources, meta={"category": "test"})
+
+    assert len(calls) == expected_calls
+    assert [document.meta for document in result["documents"]] == [
+        {"file_path": f"upload/{index}.pdf", "index": index, "category": "test", "output_format": "markdown"}
+        for index in range(count)
+        if index not in bad_indices
+    ]
+    assert all(document.content == "This is extracted PDF content" for document in result["documents"])
+    assert result["image_documents"] == []
+    for index in bad_indices:
+        assert f"Skipping PDF upload/{index}.pdf" in caplog.text
+
+
+@pytest.mark.parametrize("count", [1, 4])
+def test_converter_all_failed_batches_leave_no_images(count, tmp_path, _mock_opendataloader, monkeypatch):
+    convert = converter_module.opendataloader_pdf.convert
+
+    def fail_after_writing_output(input_path, output_dir, **kwargs):
+        convert(input_path, output_dir, **kwargs)
+        message = "corrupt PDF"
+        raise RuntimeError(message)
+
+    monkeypatch.setattr(converter_module.opendataloader_pdf, "convert", fail_after_writing_output)
+    image_output_dir = tmp_path / "images"
+    converter = OpenDataLoaderConverter(skip_failed_pdfs=True, extract_images=True, image_output_dir=image_output_dir)
+    result = converter.run(sources=[ByteStream(data=b"%PDF fake pdf") for _ in range(count)])
+
+    assert result == {"documents": [], "image_documents": []}
+    assert not list(image_output_dir.rglob("*.png"))
+
+
+@pytest.mark.parametrize("skip_failed_pdfs", [False, True])
+def test_converter_missing_output(skip_failed_pdfs, _mock_opendataloader, monkeypatch, caplog):
+    convert = converter_module.opendataloader_pdf.convert
+    calls = []
+
+    def missing_output(input_path, output_dir, **kwargs):
+        calls.append([Path(path).stem for path in input_path])
+        convert(input_path, output_dir, **kwargs)
+        if "document_1" in calls[-1]:
+            (Path(output_dir) / "document_1.md").unlink()
+
+    monkeypatch.setattr(converter_module.opendataloader_pdf, "convert", missing_output)
+    sources = [ByteStream(data=b"%PDF fake pdf") for _ in range(2)]
+    converter = OpenDataLoaderConverter(skip_failed_pdfs=skip_failed_pdfs)
+    if skip_failed_pdfs:
+        result = converter.run(sources=sources)
+        assert [document.meta["file_path"] for document in result["documents"]] == ["document_0.pdf"]
+        assert calls == [["document_0", "document_1"], ["document_0"], ["document_1"]]
+        assert "Skipping PDF document_1.pdf" in caplog.text
+    else:
+        with pytest.raises(RuntimeError, match="did not create expected file"):
+            converter.run(sources=sources)
+        assert calls == [["document_0", "document_1"]]
+
+
 def test_converter_serialization(tmp_path):
     image_output_dir = tmp_path / "images"
     converter = OpenDataLoaderConverter(
         output_format="text",
         convert_kwargs={"hybrid": "docling-fast"},
         extract_images=True,
         image_output_dir=image_output_dir,
+        skip_failed_pdfs=True,
     )
     data = converter.to_dict()
     restored = OpenDataLoaderConverter.from_dict(data)
@@ -445,6 +596,7 @@ def test_converter_serialization(tmp_path):
     assert restored.convert_kwargs["hybrid"] == "docling-fast"
     assert restored.extract_images is True
     assert restored.image_output_dir == image_output_dir
+    assert restored.skip_failed_pdfs is True
 
 
 @pytest.mark.integration
@@ -463,6 +615,29 @@ def test_real_pdf_conversion():
     assert document.meta["output_format"] == "markdown"
 
 
+@pytest.mark.integration
+@pytest.mark.parametrize("extract_images", [False, True])
+def test_real_pdf_conversion_skips_corrupt_pdf(tmp_path, extract_images):
+    good_pdf = Path(__file__).parent / "test_files" / ("pdf_with_image.pdf" if extract_images else "hello_world.pdf")
+    bad_pdf = tmp_path / "corrupt.pdf"
+    bad_pdf.write_bytes(b"not a PDF")
+    converter = OpenDataLoaderConverter(
+        skip_failed_pdfs=True,
+        extract_images=extract_images,
+        image_output_dir=tmp_path / "images" if extract_images else None,
+    )
+
+    result = converter.run(sources=[good_pdf, bad_pdf])
+
+    assert [document.meta["file_path"] for document in result["documents"]] == [good_pdf.name]
+    assert result["documents"][0].content
+    if extract_images:
+        assert result["image_documents"]
+        assert all(Path(document.meta["file_path"]).exists() for document in result["image_documents"])
+    else:
+        assert result["image_documents"] == []
+
+
 @pytest.mark.integration
 def test_real_pdf_conversion_extracts_images(tmp_path):
     pdf_file = Path(__file__).parent / "test_files" / "pdf_with_image.pdf"