Skip to content

⚡ Optimize JSON Lines (_parse_jsonl) bulk loading #118

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
wjohns989 wants to merge 1 commit into
base: main
Choose a base branch
from
Open

⚡ Optimize JSON Lines (_parse_jsonl) bulk loading #118

Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
55 changes: 41 additions & 14 deletions muninn/ingestion/parser.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -11,7 +11,7 @@
import sqlite3
from html.parser import HTMLParser
from pathlib import Path
from typing import Dict, List, Tuple
from typing import List, Tuple
from urllib.parse import quote

from muninn.ingestion.models import IngestionChunk
Expand Down Expand Up @@ -189,24 +189,43 @@ def _parse_json(path: Path) -> str:


def _parse_jsonl(path: Path) -> str:
raw_content = path.read_text(encoding="utf-8", errors="replace").strip()

@chatgpt-codex-connector chatgpt-codex-connector Bot May 13, 2026

Copy link
Copy Markdown

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

P1 Badge Stream JSONL input instead of loading whole file

Avoid reading the entire JSONL file into memory before parsing. path.read_text(...).strip() forces a full copy of the source, and the fast path then allocates another large bulk_json string; for large chat/session exports this can multiply peak memory and raise MemoryError or kill the ingestion process. The previous implementation streamed line-by-line and did not have this failure mode, so this change introduces a scalability regression in _parse_jsonl.

Useful? React with 👍 / 👎.

if not raw_content:
return ""

raw_lines = [line.strip() for line in raw_content.splitlines() if line.strip()]
if not raw_lines:
return ""

lines: List[str] = []
with path.open("r", encoding="utf-8", errors="replace") as handle:
for raw in handle:
raw_line = raw.strip()
if not raw_line:
continue
try:
payload = json.loads(raw_line)
except json.JSONDecodeError:
lines.append(raw_line)
continue

# Fast path: bulk parse
bulk_json = "[" + ",".join(raw_lines) + "]"
try:
payloads = json.loads(bulk_json)
for payload in payloads:
extracted: List[str] = []
_extract_chat_lines(payload, extracted)
if extracted:
lines.extend(extracted)
else:
lines.append(json.dumps(payload, ensure_ascii=False, sort_keys=True))
return _truncate_output("\n".join(lines))
except json.JSONDecodeError:
pass

# Fallback path: loop
for raw_line in raw_lines:
try:
payload = json.loads(raw_line)
extracted = []
_extract_chat_lines(payload, extracted)
if extracted:
lines.extend(extracted)
else:
lines.append(json.dumps(payload, ensure_ascii=False, sort_keys=True))
except json.JSONDecodeError:
lines.append(raw_line)

return _truncate_output("\n".join(lines))
Comment on lines +192 to 230

@gemini-code-assist gemini-code-assist Bot May 13, 2026

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

medium

The _parse_jsonl implementation introduces a significant memory regression and contains duplicated logic.

  1. Memory Efficiency: By using path.read_text().splitlines(), the entire file is loaded into memory as a string, then as a list of strings, and finally joined into a third large string for bulk parsing. For a 100MB file (the maximum allowed), this can result in a ~300MB peak memory spike per worker. Iterating over the file handle directly to build raw_lines avoids the first intermediate string spike.
  2. Maintainability: The logic for processing a JSON payload (extracting chat lines or falling back to json.dumps) is duplicated between the fast path and the fallback path. Refactoring this into a local helper function improves clarity and ensures consistency.

This refactor aligns with the SOTA+ philosophy of 'Production-Grade' and 'Quality Over Speed' by optimizing resource usage while improving code structure.

    with path.open("r", encoding="utf-8", errors="replace") as handle:
        raw_lines = [line.strip() for line in handle if line.strip()]

    if not raw_lines:
        return ""

    lines: List[str] = []

    def process_payload(payload):
        extracted: List[str] = []
        _extract_chat_lines(payload, extracted)
        if extracted:
            lines.extend(extracted)
        else:
            lines.append(json.dumps(payload, ensure_ascii=False, sort_keys=True))

    # Fast path: bulk parse
    bulk_json = "[" + ",".join(raw_lines) + "]"
    try:
        payloads = json.loads(bulk_json)
        for payload in payloads:
            process_payload(payload)
        return _truncate_output("\n".join(lines))
    except json.JSONDecodeError:
        pass

    # Fallback path: line-by-line loop
    for raw_line in raw_lines:
        try:
            payload = json.loads(raw_line)
            process_payload(payload)
        except json.JSONDecodeError:
            lines.append(raw_line)

    return _truncate_output("\n".join(lines))
References
  1. Adherence to SOTA+ standards of precision and quality, ensuring production-grade robustness and optimal resource usage. (link)


Expand All @@ -232,12 +251,14 @@ def _parse_html(path: Path) -> str:
def _parse_pdf(path: Path) -> str:
"""Parse a PDF file via the subprocess sandbox for process isolation (Phase 17)."""
from muninn.ingestion.sandbox import sandboxed_parse_binary

return sandboxed_parse_binary(path, "pdf", timeout=30.0)


def _parse_docx(path: Path) -> str:
"""Parse a DOCX file via the subprocess sandbox for process isolation (Phase 17)."""
from muninn.ingestion.sandbox import sandboxed_parse_binary

return sandboxed_parse_binary(path, "docx", timeout=30.0)


Expand All @@ -258,7 +279,10 @@ def _parse_sqlite(path: Path) -> str:
preferred = [
name
for name in table_names
if any(token in name.lower() for token in ("chat", "conversation", "message", "session", "copilot", "ai", "prompt"))
if any(
token in name.lower()
for token in ("chat", "conversation", "message", "session", "copilot", "ai", "prompt")
)
]
ordered_tables = preferred + [name for name in table_names if name not in preferred]
ordered_tables = ordered_tables[:12]
Expand Down Expand Up @@ -298,7 +322,10 @@ def _parse_sqlite(path: Path) -> str:
if len(value_text) > 2000:
value_text = value_text[:2000] + "..."
key_lower = str(key).lower()
if any(token in key_lower for token in ("content", "text", "prompt", "response", "message", "body", "value")):
if any(
token in key_lower
for token in ("content", "text", "prompt", "response", "message", "body", "value")
):
fallback_texts.append(f"[{key}] {value_text}")
if fallback_texts:
lines.extend(fallback_texts)
Expand Down Expand Up @@ -412,4 +439,4 @@ def build_chunks(
},
)
)
return chunks
return chunks
Loading