TIMX 739 - lxml memory pointer bug

tests/sources/xml/test_marc.py

@@ -2282,6 +2282,11 @@ def test_marc_parse_source_record_recovers_from_malformed_xml():
     assert Marc.find_element_string(source_record, "leader") == "x"
 
 
+def test_marc_parse_source_record_raises_parsing_errors():
+    with pytest.raises(etree.XMLSyntaxError, match="Document is empty"):
+        Marc.parse_source_record(b"")
+
+
 def test_record_is_deleted_returns_true_if_deleted():
     deleted_record = Marc.parse_source_file("tests/fixtures/marc/marc_record_deleted.xml")
     assert Marc.record_is_deleted(next(deleted_record)) is True

tests/test_cli.py

@@ -213,7 +213,7 @@ def test_transform_run_timestamp_argument_not_passed_and_timestamp_minted(
         assert transformer.run_data["run_date"] == "2024-06-03"
 
 
-def test_transform_no_memory_fault_for_threaded_bs4_parsing(monkeypatch, tmp_path):
+def test_transform_no_memory_fault_for_bs4_parsing(monkeypatch, tmp_path):
     """This test requires running the CLI as a subprocess to simulate the 'uv run ...'
     context. In this context, we have observed memory faults when BeautifulSoup4 is used
     by a source, and the number of records in the run requires multiple batches during
@@ -238,6 +238,39 @@ def test_transform_no_memory_fault_for_threaded_bs4_parsing(monkeypatch, tmp_pat
     assert result.returncode == 0
 
 
+def test_transform_no_memory_fault_for_marc_parsing(tmp_path):
+    """As test_transform_no_memory_fault_for_bs4_parsing, for lxml parsing.
+
+    Requires one more record than a single parquet dataset write batch.
+    """
+    record = (
+        "<record><leader>03282nam  2200721Ki 4500</leader>"
+        '<controlfield tag="001">990027185640106761</controlfield>'
+        '<controlfield tag="008">170906s2016    fr mun| o         e zxx d</controlfield>'
+        '<datafield tag="245"><subfield code="a">Title</subfield></datafield></record>'
+    )
+    input_file = tmp_path / "alma-2026-01-01-daily-extracted-records-to-index.xml"
+    input_file.write_text(f"<collection>{record * 1001}</collection>")
+
+    result = subprocess.run(  # noqa: S603
+        [  # noqa: S607
+            "uv",
+            "run",
+            "transform",
+            "-s",
+            "alma",
+            "-i",
+            str(input_file),
+            "-o",
+            f"{tmp_path}/dataset",
+        ],
+        capture_output=True,
+        text=True,
+        check=False,
+    )
+    assert result.returncode == 0
+
+
 def test_transform_critical_error_prevents_writing(
     caplog, runner, source_input_file, source_transformer, empty_dataset_location
 ):

transmogrifier/sources/xmltransformer.py

@@ -1,6 +1,7 @@
 from __future__ import annotations
 
-import threading
+import itertools
+from concurrent.futures import ThreadPoolExecutor
 from typing import TYPE_CHECKING
 
 import smart_open  # type: ignore[import-untyped]
@@ -12,6 +13,8 @@
 if TYPE_CHECKING:
     from collections.abc import Iterator
 
+READ_CHUNK_SIZE = 100
+
 
 class XMLTransformer(Transformer):
     """XML transformer class."""
@@ -21,9 +24,25 @@ def read_source_file(cls, source_file: str) -> Iterator[bytes]:
         """
         Read XML file and yield each <record> element as bytes.
 
+        The file is read on a single, dedicated thread.  pyarrow may resume this generator
+        under a new thread state each time, and lxml has a string dictionary per thread
+        state.  If iterparse finishes under a different thread state than it started, one
+        that has parsed something else, lxml replaces the document's dictionary and the
+        application aborts with a memory fault when that document is freed.
+
         Args:
             source_file: A file containing source records to be transformed.
         """
+        records = cls._iterparse_source_file(source_file)
+        with ThreadPoolExecutor(max_workers=1) as executor:
+            while chunk := executor.submit(
+                lambda: list(itertools.islice(records, READ_CHUNK_SIZE))
+            ).result():
+                yield from chunk
+
+    @classmethod
+    def _iterparse_source_file(cls, source_file: str) -> Iterator[bytes]:
+        """Iteratively parse an XML file and yield each <record> element as bytes."""
         with smart_open.open(source_file, "rb") as file:
             for _, element in etree.iterparse(
                 file,
@@ -42,18 +61,7 @@ def parse_source_record(cls, source_record: bytes) -> Tag:
         Args:
             source_record: An XML source record as bytes.
         """
-        return cls.parse_bs4_in_isolated_thread(source_record)
-
-    @classmethod
-    def parse_bs4_in_isolated_thread(cls, source_record: bytes) -> Tag:
-        def parse() -> Tag:
-            nonlocal source_record
-            source_record = BeautifulSoup(source_record, "xml")
-
-        thread = threading.Thread(target=parse)
-        thread.start()
-        thread.join()
-        return source_record
+        return BeautifulSoup(source_record, "xml")
 
     @classmethod
     def get_main_titles(cls, _source_record: Tag) -> list[Tag]: