Collapse docling compression to a single function

This commit is contained in:
Yiorgis Gozadinos 2026-06-24 10:40:09 +03:00
parent e95ac2e25d
commit 2b2b475279
No known key found for this signature in database
10 changed files with 29 additions and 57 deletions

View file

@ -4,6 +4,7 @@
### Fixed ### Fixed
- zstd compression uses a fresh `zstandard` compressor/decompressor per call instead of shared module-level singletons, fixing process segfaults when ingester workers compress documents concurrently (Python < 3.14). - zstd compression uses a fresh `zstandard` compressor/decompressor per call instead of shared module-level singletons, fixing process segfaults when ingester workers compress documents concurrently (Python < 3.14).
- Ingestion compresses the DoclingDocument structure directly from the in-memory dict, removing one full-size serialized copy from per-document peak memory.
## [0.61.0] - 2026-06-23 ## [0.61.0] - 2026-06-23

View file

@ -643,7 +643,7 @@ def _apply_descriptions_sync(
pic.meta = PictureMeta() pic.meta = PictureMeta()
pic.meta.description = DescriptionMetaField(text=text) pic.meta.description = DescriptionMetaField(text=text)
structure_bytes, _ = compress_docling_split(docling_doc.model_dump_json()) structure_bytes, _ = compress_docling_split(docling_doc.model_dump(mode="json"))
doc.docling_document = structure_bytes doc.docling_document = structure_bytes
doc.docling_version = docling_doc.version doc.docling_version = docling_doc.version
return len(descriptions) return len(descriptions)

View file

@ -34,20 +34,7 @@ def decompress_json(data: bytes) -> str:
return _zstd_decompress(data).decode("utf-8") return _zstd_decompress(data).decode("utf-8")
def compress_docling_split(json_str: str) -> tuple[bytes, bytes | None]: def compress_docling_split(data: dict) -> tuple[bytes, bytes | None]:
"""Parse a DoclingDocument JSON string and compress it.
Thin wrapper over :func:`compress_docling_data` for callers that only hold
the serialized string store migrations and rebuild-from-blob, neither of
which is speed-sensitive. The ingestion hot path should call
``compress_docling_data`` with ``DoclingDocument.model_dump(mode="json")``
instead, to avoid serializing the document to a full JSON string only to
parse it straight back into a dict.
"""
return compress_docling_data(json.loads(json_str))
def compress_docling_data(data: dict) -> tuple[bytes, bytes | None]:
"""Split a DoclingDocument dict into structure and pages, compress both with zstd. """Split a DoclingDocument dict into structure and pages, compress both with zstd.
Picture image URIs are stripped from the structure blob they are stored on Picture image URIs are stripped from the structure blob they are stored on
@ -70,10 +57,7 @@ def compress_docling_data(data: dict) -> tuple[bytes, bytes | None]:
if isinstance(picture, dict): if isinstance(picture, dict):
picture["image"] = None picture["image"] = None
structure_bytes = _zstd_compress(json.dumps(data).encode("utf-8")) structure_bytes = compress_json(json.dumps(data))
pages_bytes = compress_json(json.dumps(pages)) if pages else None
pages_bytes = None
if pages:
pages_bytes = _zstd_compress(json.dumps(pages).encode("utf-8"))
return structure_bytes, pages_bytes return structure_bytes, pages_bytes

View file

@ -4,7 +4,7 @@ from typing import TYPE_CHECKING
from pydantic import BaseModel, Field from pydantic import BaseModel, Field
from haiku.rag.store.compression import compress_docling_data, decompress_json from haiku.rag.store.compression import compress_docling_split, decompress_json
if TYPE_CHECKING: if TYPE_CHECKING:
from docling_core.types.doc.document import DoclingDocument, PageItem from docling_core.types.doc.document import DoclingDocument, PageItem
@ -32,7 +32,7 @@ class Document(BaseModel):
Sets docling_document (zstd-compressed structure without pages), Sets docling_document (zstd-compressed structure without pages),
docling_pages (zstd-compressed page images), and docling_version. docling_pages (zstd-compressed page images), and docling_version.
""" """
structure, pages = compress_docling_data(docling_doc.model_dump(mode="json")) structure, pages = compress_docling_split(docling_doc.model_dump(mode="json"))
self.docling_document = structure self.docling_document = structure
self.docling_pages = pages self.docling_pages = pages
self.docling_version = docling_doc.version self.docling_version = docling_doc.version

View file

@ -58,7 +58,7 @@ async def _apply_split_pages_zstd(store: Store) -> None: # pragma: no cover
json_str = docling_blob.decode("utf-8") json_str = docling_blob.decode("utf-8")
# Split structure and pages, re-compress with zstd # Split structure and pages, re-compress with zstd
structure_bytes, pages_bytes = compress_docling_split(json_str) structure_bytes, pages_bytes = compress_docling_split(json.loads(json_str))
metadata_raw = row.get("metadata") metadata_raw = row.get("metadata")
metadata_str = ( metadata_str = (

View file

@ -54,8 +54,7 @@ class TestDoclingCompressionSplit:
"texts": [{"text": "hello"}], "texts": [{"text": "hello"}],
"pages": {"1": {"image": "base64data"}, "2": {"image": "more"}}, "pages": {"1": {"image": "base64data"}, "2": {"image": "more"}},
} }
json_str = json.dumps(data) structure_bytes, pages_bytes = compress_docling_split(data)
structure_bytes, pages_bytes = compress_docling_split(json_str)
assert structure_bytes is not None assert structure_bytes is not None
assert pages_bytes is not None assert pages_bytes is not None
@ -73,8 +72,7 @@ class TestDoclingCompressionSplit:
def test_split_without_pages(self): def test_split_without_pages(self):
data = {"name": "test_doc", "texts": []} data = {"name": "test_doc", "texts": []}
json_str = json.dumps(data) structure_bytes, pages_bytes = compress_docling_split(data)
structure_bytes, pages_bytes = compress_docling_split(json_str)
assert structure_bytes is not None assert structure_bytes is not None
assert pages_bytes is None assert pages_bytes is None
@ -84,8 +82,7 @@ class TestDoclingCompressionSplit:
def test_split_with_empty_pages(self): def test_split_with_empty_pages(self):
data = {"name": "test_doc", "texts": [], "pages": {}} data = {"name": "test_doc", "texts": [], "pages": {}}
json_str = json.dumps(data) structure_bytes, pages_bytes = compress_docling_split(data)
structure_bytes, pages_bytes = compress_docling_split(json_str)
assert structure_bytes is not None assert structure_bytes is not None
assert pages_bytes is None assert pages_bytes is None

View file

@ -397,8 +397,7 @@ class TestDocumentItemMigration:
from haiku.rag.store.upgrades.v0_40_0 import _apply_populate_document_items from haiku.rag.store.upgrades.v0_40_0 import _apply_populate_document_items
docling_doc = _make_docling_doc() docling_doc = _make_docling_doc()
json_str = docling_doc.model_dump_json() structure, pages = compress_docling_split(docling_doc.model_dump(mode="json"))
structure, pages = compress_docling_split(json_str)
# Create a database at a pre-migration version with a document # Create a database at a pre-migration version with a document
async with Store(temp_db_path, create=True, skip_migration_check=True) as store: async with Store(temp_db_path, create=True, skip_migration_check=True) as store:
@ -687,7 +686,7 @@ class TestCompressDoclingSplitStripsPictureUris:
"pages": {}, "pages": {},
} }
structure_bytes, pages_bytes = compress_docling_split(json.dumps(doc_json)) structure_bytes, pages_bytes = compress_docling_split(doc_json)
decoded = json.loads(decompress_json(structure_bytes)) decoded = json.loads(decompress_json(structure_bytes))
for pic in decoded["pictures"]: for pic in decoded["pictures"]:

View file

@ -35,7 +35,7 @@ async def test_populate_handles_extra_columns_on_items_table(temp_db_path):
accepted even though it only writes the original 6 columns. accepted even though it only writes the original 6 columns.
""" """
docling_doc = _simple_docling_doc() docling_doc = _simple_docling_doc()
structure, pages = compress_docling_split(docling_doc.model_dump_json()) structure, pages = compress_docling_split(docling_doc.model_dump(mode="json"))
async with Store(temp_db_path, create=True, skip_migration_check=True) as store: async with Store(temp_db_path, create=True, skip_migration_check=True) as store:
# _init_tables already created document_items with the latest schema. # _init_tables already created document_items with the latest schema.

View file

@ -26,7 +26,7 @@ class TestV0_48_0Migration:
async def test_backfill_populates_levels(self, temp_db_path): async def test_backfill_populates_levels(self, temp_db_path):
docling_doc = _docling_with_levels() docling_doc = _docling_with_levels()
structure, pages = compress_docling_split(docling_doc.model_dump_json()) structure, pages = compress_docling_split(docling_doc.model_dump(mode="json"))
async with Store(temp_db_path, create=True, skip_migration_check=True) as store: async with Store(temp_db_path, create=True, skip_migration_check=True) as store:
await store.set_haiku_version("0.45.0") await store.set_haiku_version("0.45.0")
@ -79,7 +79,7 @@ class TestV0_48_0Migration:
async def test_backfill_idempotent(self, temp_db_path): async def test_backfill_idempotent(self, temp_db_path):
docling_doc = _docling_with_levels() docling_doc = _docling_with_levels()
structure, pages = compress_docling_split(docling_doc.model_dump_json()) structure, pages = compress_docling_split(docling_doc.model_dump(mode="json"))
async with Store(temp_db_path, create=True, skip_migration_check=True) as store: async with Store(temp_db_path, create=True, skip_migration_check=True) as store:
await store.set_haiku_version("0.45.0") await store.set_haiku_version("0.45.0")

View file

@ -281,14 +281,11 @@ def test_set_docling_with_page_images():
assert "1" in pages assert "1" in pages
def test_set_docling_dict_path_matches_string_path(): def test_compress_docling_split_dict_sources_match():
"""The ingestion dict path must produce the same stored bytes as the """Both ways of building the compression input dict must yield identical
legacy string path. stored bytes: ``model_dump(mode="json")`` (used at ingest) and
``json.loads(model_dump_json())`` (used when migrating a stored string).
``set_docling`` feeds ``compress_docling_data`` the dict from In particular int-keyed ``pages`` must serialize to string keys either way.
``model_dump(mode="json")`` instead of round-tripping through
``model_dump_json()`` + ``json.loads``. The on-disk format must not change
in particular int-keyed ``pages`` must still serialize to string keys.
""" """
import json import json
@ -296,28 +293,22 @@ def test_set_docling_dict_path_matches_string_path():
from docling_core.types.doc.document import DoclingDocument, PageItem from docling_core.types.doc.document import DoclingDocument, PageItem
from docling_core.types.doc.labels import DocItemLabel from docling_core.types.doc.labels import DocItemLabel
from haiku.rag.store.compression import ( from haiku.rag.store.compression import compress_docling_split
compress_docling_data,
compress_docling_split,
decompress_json,
)
docling_doc = DoclingDocument(name="equivalence_test") docling_doc = DoclingDocument(name="equivalence_test")
docling_doc.add_text(label=DocItemLabel.PARAGRAPH, text="Hello world") docling_doc.add_text(label=DocItemLabel.PARAGRAPH, text="Hello world")
docling_doc.pages[1] = PageItem(size=Size(width=612, height=792), page_no=1) docling_doc.pages[1] = PageItem(size=Size(width=612, height=792), page_no=1)
struct_dict, pages_dict = compress_docling_data( struct_dump, pages_dump = compress_docling_split(
docling_doc.model_dump(mode="json") docling_doc.model_dump(mode="json")
) )
struct_str, pages_str = compress_docling_split(docling_doc.model_dump_json()) struct_str, pages_str = compress_docling_split(
json.loads(docling_doc.model_dump_json())
)
assert json.loads(decompress_json(struct_dict)) == json.loads( assert struct_dump == struct_str
decompress_json(struct_str) assert pages_dump is not None and pages_str is not None
) assert pages_dump == pages_str
assert pages_dict is not None and pages_str is not None
assert json.loads(decompress_json(pages_dict)) == json.loads(
decompress_json(pages_str)
)
def test_get_page_images(): def test_get_page_images():