Reduce CSV conversion memory while preserving upstream output

backendN/app/converter.py

@@ -55,6 +55,9 @@ def __init__(self) -> None:
             self._import_error = str(err)
             return
         self._md = MarkItDown(enable_plugins=False)
+        from app.csv_converter import register_csv_converter
+
+        register_csv_converter(self._md)
         self._stream_info = StreamInfo
         self._import_error = ""
 

backendN/app/csv_converter.py

@@ -0,0 +1,90 @@
+"""Row-at-a-time CSV formatting for verified MarkItDown versions.
+
+Keep decoding, cell escaping and table semantics identical to upstream.
+Two CSV passes determine width/blank boundaries without retaining all rows.
+The input text and complete output still occupy memory; this is not a
+streaming HTTP/Storage contract or a bound on total parser memory.
+"""
+
+from __future__ import annotations
+
+import csv
+import io
+from importlib.metadata import version
+
+from markitdown import DocumentConverter, DocumentConverterResult
+
+VERIFIED_VERSIONS = frozenset({"0.1.8", "0.1.9"})
+
+
+class RowStreamingCsvConverter(DocumentConverter):
+    def __init__(self):
+        # The private escape helper is deliberately version-gated below. Reuse
+        # it so pipes, backslash runs and embedded newlines remain identical.
+        from markitdown.converters import _csv_converter as upstream
+
+        self._upstream = upstream.CsvConverter()
+        self._escape = upstream._escape_table_cell
+        self._detect = upstream.from_bytes
+
+    def accepts(self, file_stream, stream_info, **kwargs):
+        return self._upstream.accepts(file_stream, stream_info, **kwargs)
+
+    def convert(self, file_stream, stream_info, **kwargs):
+        if stream_info.charset:
+            content = file_stream.read().decode(stream_info.charset)
+        else:
+            data = file_stream.read()
+            detected = self._detect(data).best()
+            content = str(detected) if detected is not None else data.decode("utf-8", errors="ignore")
+            del data, detected
+
+        # Upstream strips BOMs and preserves embedded CR/LF until cell escaping.
+        source = io.StringIO(content.lstrip("\ufeff"), newline="")
+        del content
+        with source, io.StringIO() as output:
+            width = 0
+            header = first_data = last_data = None
+            for number, row in enumerate(csv.reader(source)):
+                if not row:
+                    continue
+                width = max(width, len(row))
+                if header is None:
+                    header = number
+                elif first_data is None:
+                    first_data = number
+                last_data = number
+            if header is None:
+                return DocumentConverterResult(markdown="")
+
+            source.seek(0)
+            for number, row in enumerate(csv.reader(source)):
+                if number < header:
+                    continue
+                if number > last_data:
+                    break
+                # Leading/trailing empty rows and the empty run immediately
+                # after the header are trimmed. Interior empty rows survive.
+                if number != header and (first_data is None or number < first_data):
+                    continue
+                row.extend([""] * (width - len(row)))
+                if number != header:
+                    output.write("\n")
+                output.write("| " + " | ".join(self._escape(cell) for cell in row) + " |")
+                if number == header:
+                    output.write("\n| " + " | ".join(["---"] * width) + " |")
+            return DocumentConverterResult(markdown=output.getvalue())
+
+
+def register_csv_converter(engine) -> bool:
+    """Unknown/older upstream versions keep their own converter unchanged."""
+    if version("markitdown") not in VERIFIED_VERSIONS:
+        return False
+    try:
+        converter = RowStreamingCsvConverter()
+    except (ImportError, AttributeError):
+        return False
+    # MarkItDown's public registration API tries the latest converter first
+    # at the default specific-format priority. Detection/fallback stay intact.
+    engine.register_converter(converter)
+    return True

backendN/tests/test_csv_capacity.py

@@ -0,0 +1,116 @@
+"""CSV output must match the installed, verified upstream converter exactly."""
+import csv
+import io
+import random
+from importlib.metadata import version
+from pathlib import Path
+from types import SimpleNamespace
+
+import pytest
+from markitdown import StreamInfo
+from markitdown.converters import CsvConverter
+
+from app import csv_converter
+
+VERIFIED = version("markitdown") in csv_converter.VERIFIED_VERSIONS
+optimized = pytest.mark.skipif(not VERIFIED, reason="unverified upstream retains its original converter")
+
+
+def compare(data, charset="utf-8"):
+    info = StreamInfo(extension=".csv", charset=charset)
+    actual = csv_converter.RowStreamingCsvConverter().convert(io.BytesIO(data), info).markdown
+    expected = CsvConverter().convert(io.BytesIO(data), info).markdown
+    assert actual == expected
+    return actual
+
+
+@optimized
+@pytest.mark.parametrize("text", [
+    "", "\n\r\n\n", '""\n', "A,B", "\n\nA,B\n\n\n",
+    "\n\nA,B\n\n1,2,3\n\n\n4\n\n", "A\n1,2,3\n4,5",
+    "A,B\r\n1,2\r3,4\n", '\ufeff\ufeffA,B\n1,2',
+    '"A|B",C\n"line\nbreak","a\\\\|b"\n"carriage\rreturn","quote""quote"',
+    "名前,Größe,Español,🙂\n日本語,groß,café,終わり",
+    "A,B\n,,\n\n1,2\n\n,,\n", "A\n" + "\n" * 2000 + "last\n" * 2,
+])
+def test_boundary_rows_and_cells_match_upstream(text):
+    compare(text.encode())
+
+
+@optimized
+def test_random_ragged_quoted_tables_match_upstream():
+    rng = random.Random(428)
+    values = ["", "plain", "漢字🙂", "a|b", "x\\|y", 'quote"value', "first\nsecond", "CR\rLF\r\n", " café "]
+    for _ in range(100):
+        source = io.StringIO(newline="")
+        writer = csv.writer(source)
+        writer.writerows([[rng.choice(values) for _ in range(rng.randrange(7))] for _ in range(rng.randrange(30))])
+        compare(source.getvalue().encode())
+
+
+@optimized
+@pytest.mark.parametrize("charset", ["utf-8", "utf-16", "cp1252", None])
+def test_explicit_and_detected_encodings_match_upstream(charset):
+    text = 'Prénom,City\nAndré,Zürich\n'
+    compare(text.encode(charset or "utf-8"), charset)
+
+
+@optimized
+def test_undetectable_bytes_keep_upstream_utf8_fallback(monkeypatch):
+    from markitdown.converters import _csv_converter as upstream
+    monkeypatch.setattr(upstream, "from_bytes", lambda data: SimpleNamespace(best=lambda: None))
+    compare(b'A,B\n\xff,1', None)
+
+
+@optimized
+def test_parser_failure_and_borrowed_stream_ownership_match_upstream():
+    data = ('A\n' + 'x' * (csv.field_size_limit() + 1)).encode()
+    for converter in [CsvConverter(), csv_converter.RowStreamingCsvConverter()]:
+        source = io.BytesIO(data)
+        with pytest.raises(csv.Error):
+            converter.convert(source, StreamInfo(extension=".csv", charset="utf-8"))
+        assert not source.closed
+
+
+def test_unknown_version_keeps_engine_registration_untouched(monkeypatch):
+    monkeypatch.setattr(csv_converter, "version", lambda name: "0.1.10")
+    engine = SimpleNamespace(register_converter=lambda *args, **kwargs: pytest.fail("unknown converter registered"))
+    assert csv_converter.register_csv_converter(engine) is False
+
+
+def test_unverified_version_keeps_actual_upstream_csv_output(monkeypatch):
+    from app.converter import NormalConverter
+    monkeypatch.setattr(csv_converter, "version", lambda name: "0.1.10")
+    data = b'Name\n"Ada|Lovelace",99\nfinal,100'
+    expected = CsvConverter().convert(io.BytesIO(data), StreamInfo(extension=".csv", charset="utf-8")).markdown.strip()
+    actual, warning = NormalConverter().convert(data, "csv", filename="old.csv", title="old")
+    assert actual == expected
+    assert warning is None
+
+
+@optimized
+def test_missing_upstream_helper_keeps_original_converter(monkeypatch):
+    from markitdown.converters import _csv_converter as upstream
+    monkeypatch.delattr(upstream, "_escape_table_cell")
+    engine = SimpleNamespace(register_converter=lambda *args, **kwargs: pytest.fail("incomplete converter registered"))
+    assert csv_converter.register_csv_converter(engine) is False
+
+
+def test_document_pools_share_identical_csv_implementation():
+    root = Path(__file__).resolve().parents[2]
+    assert (root / "backendN/app/csv_converter.py").read_bytes() == (root / "backendZ/app/csv_converter.py").read_bytes()
+
+
+@optimized
+def test_normal_route_uses_optimized_converter_with_unchanged_output(convert, monkeypatch):
+    calls = []
+    original = csv_converter.RowStreamingCsvConverter.convert
+    def observed(self, *args, **kwargs):
+        calls.append(True)
+        return original(self, *args, **kwargs)
+    monkeypatch.setattr(csv_converter.RowStreamingCsvConverter, "convert", observed)
+    response = convert("ragged.csv", b'Name\n"Ada|Lovelace",99\n"end-marker",100')
+    assert response.status_code == 200, response.text
+    assert calls
+    assert "end-marker" in response.json()["content"]
+    assert r"Ada\|Lovelace" in response.json()["content"]

backendZ/app/converter.py

@@ -38,6 +38,9 @@ def __init__(self) -> None:
             self._md, self._stream_info, self.error = None, None, str(err)
             return
         self._md = MarkItDown(enable_plugins=False)
+        from app.csv_converter import register_csv_converter
+
+        register_csv_converter(self._md)
         self._stream_info = StreamInfo
         self.error = ""
 

backendZ/app/csv_converter.py

@@ -0,0 +1,90 @@
+"""Row-at-a-time CSV formatting for verified MarkItDown versions.
+
+Keep decoding, cell escaping and table semantics identical to upstream.
+Two CSV passes determine width/blank boundaries without retaining all rows.
+The input text and complete output still occupy memory; this is not a
+streaming HTTP/Storage contract or a bound on total parser memory.
+"""
+
+from __future__ import annotations
+
+import csv
+import io
+from importlib.metadata import version
+
+from markitdown import DocumentConverter, DocumentConverterResult
+
+VERIFIED_VERSIONS = frozenset({"0.1.8", "0.1.9"})
+
+
+class RowStreamingCsvConverter(DocumentConverter):
+    def __init__(self):
+        # The private escape helper is deliberately version-gated below. Reuse
+        # it so pipes, backslash runs and embedded newlines remain identical.
+        from markitdown.converters import _csv_converter as upstream
+
+        self._upstream = upstream.CsvConverter()
+        self._escape = upstream._escape_table_cell
+        self._detect = upstream.from_bytes
+
+    def accepts(self, file_stream, stream_info, **kwargs):
+        return self._upstream.accepts(file_stream, stream_info, **kwargs)
+
+    def convert(self, file_stream, stream_info, **kwargs):
+        if stream_info.charset:
+            content = file_stream.read().decode(stream_info.charset)
+        else:
+            data = file_stream.read()
+            detected = self._detect(data).best()
+            content = str(detected) if detected is not None else data.decode("utf-8", errors="ignore")
+            del data, detected
+
+        # Upstream strips BOMs and preserves embedded CR/LF until cell escaping.
+        source = io.StringIO(content.lstrip("\ufeff"), newline="")
+        del content
+        with source, io.StringIO() as output:
+            width = 0
+            header = first_data = last_data = None
+            for number, row in enumerate(csv.reader(source)):
+                if not row:
+                    continue
+                width = max(width, len(row))
+                if header is None:
+                    header = number
+                elif first_data is None:
+                    first_data = number
+                last_data = number
+            if header is None:
+                return DocumentConverterResult(markdown="")
+
+            source.seek(0)
+            for number, row in enumerate(csv.reader(source)):
+                if number < header:
+                    continue
+                if number > last_data:
+                    break
+                # Leading/trailing empty rows and the empty run immediately
+                # after the header are trimmed. Interior empty rows survive.
+                if number != header and (first_data is None or number < first_data):
+                    continue
+                row.extend([""] * (width - len(row)))
+                if number != header:
+                    output.write("\n")
+                output.write("| " + " | ".join(self._escape(cell) for cell in row) + " |")
+                if number == header:
+                    output.write("\n| " + " | ".join(["---"] * width) + " |")
+            return DocumentConverterResult(markdown=output.getvalue())
+
+
+def register_csv_converter(engine) -> bool:
+    """Unknown/older upstream versions keep their own converter unchanged."""
+    if version("markitdown") not in VERIFIED_VERSIONS:
+        return False
+    try:
+        converter = RowStreamingCsvConverter()
+    except (ImportError, AttributeError):
+        return False
+    # MarkItDown's public registration API tries the latest converter first
+    # at the default specific-format priority. Detection/fallback stay intact.
+    engine.register_converter(converter)
+    return True

backendZ/tests/test_document_containers.py

@@ -90,3 +90,24 @@ def test_real_documents_keep_conversion_content_inside_archive(converter, ext):
     assert entries[0].status == "DONE"
     assert entries[0].content == direct
     assert direct
+
+
+def test_archive_csv_preserves_upstream_table_and_uses_row_converter(converter, monkeypatch):
+    import io
+    from importlib.metadata import version
+    from markitdown import MarkItDown, StreamInfo
+    from app import csv_converter
+    if version("markitdown") not in csv_converter.VERIFIED_VERSIONS:
+        pytest.skip("unverified upstream retains its original converter")
+    calls = []
+    original = csv_converter.RowStreamingCsvConverter.convert
+    def observed(self, *args, **kwargs):
+        calls.append(True)
+        return original(self, *args, **kwargs)
+    monkeypatch.setattr(csv_converter.RowStreamingCsvConverter, "convert", observed)
+    data = '\nName\n\n"Ada|Lovelace",99\n\n"終わり🙂",100\n\n'.encode()
+    upstream = MarkItDown(enable_plugins=False).convert_stream(io.BytesIO(data), stream_info=StreamInfo(extension=".csv", mimetype="text/csv", filename="ragged.csv")).markdown.strip()
+    entries = archive.process_archive(make_zip({"ragged.csv": data}), converter.documents, ocr_available=False)
+    assert calls
+    assert entries[0].status == "DONE"
+    assert entries[0].content == upstream

docs/LLMISER_OPTIMIZATION_PLAN.md

@@ -10,6 +10,8 @@ Shared reader statistics shipped in PR #9 at `84e510db62ea41f31c131771eb7d32b5e9
 
 PR #11 released [large-preview control](LLMISER_LARGE_PREVIEWS.md) at `0501cd0381b4a41e9a9e94abf791ebd4fd8fd188`. Browser-generated dense Markdown exposed multi-second mount/layout work. Automatic formatting now pauses above 50,000 characters; full formatting is explicitly available, and raw content/exports are complete. Local checks passed 261 tests, the final 20-test selection and production build; all seven CI jobs and all six production deployments passed. The 28 October audit passed 20 HTTP checks with all pools 2/2 ready. Next work should measure native textarea input/representative devices or audit backend parser capacity; a worker alone will not eliminate the observed DOM/layout cost.
 
+The current `codex/csv-parser-memory` candidate contains [a source capacity audit and CSV allocation reduction](LLMISER_PARSER_CAPACITY.md) for N/Z. It preserves output on verified MarkItDown 0.1.8/0.1.9 and falls back unchanged on other versions. Three fresh processes per mode show lower large-fixture peak memory; no concurrency/limit changes are justified by that local evidence. See its PR for release state, then measure workbook/native and final response/merge costs before extending this slice.
+
 ## Verified starting point
 
 The copied source already contains atomic job startup/finalization, cleanup recovery, upload/active-job/storage-reservation admission, shared archive container checks, bounded PDF page uploads/OCR preparation, lazy Markdown/highlighter loading and current-edit export snapshots. Do not reimplement these based on old “proposed” checklist rows. Their deployment/capacity evidence is distinct from source tests.
@@ -40,7 +42,7 @@ Acceptance: expired link, later-part failure, complete Unicode assembly, restart
 | Work | Next bounded change / dependency |
 | --- | --- |
 | Shared statistics and large previews | Counting released in PR #9; browser profiling and large-preview control released in PR #11; measure native input/devices before deeper editor/worker changes |
-| Backend parser capacity | Audit all parsers beyond the already-shipped PDF/OCR preparation bounds; preserve fidelity and measure synchronized memory before tuning concurrency |
+| Backend parser capacity | Source audit and synchronized CSV memory slice implemented on the current branch; check its PR for release. Next measure workbook/native and response/merge memory; preserve fidelity before tuning concurrency |
 | Shared deadline | Add an optional, backward-compatible remaining-budget protocol, reserving finalization time; test slow failures/cold starts |
 | Controller detail pagination | Top-level lists already paginate; bound files/tree/events/work-items with stable cursors and truthful totals; query-plan evidence before new indexes |
 | Replay-stable object identities | Fence attempt-specific writes before deterministic IDs; SQL attempt guards alone do not prevent a stale Storage overwrite; additive rollout for new roots only |