--- title: 'TEXTIQ-392 — Chunk schema versioning (v1.0)' type: 'feature' created: '2026-04-21' status: 'done' context: - 'docs/implementation-artifacts/de-airflow-gap-audit-2026-04-20.md' jira_id: TEXTIQ-392 gap_id: GAP-043 baseline_commit: 1af75fd7d8b22493bb9d247a3e1843f93debcd7c --- ## Intent **Problem:** Hierarchical chunks written by DE carry zero version marker. Any future change to the chunk envelope (metadata shape, new fields, renames) would break downstream consumers (chatbot-service, textiq-doc-generation) silently — no way to detect or gate on version. GAP-043 in `de-airflow-gap-audit-2026-04-20.md`. **Approach:** Introduce a single constant `CHUNK_SCHEMA_VERSION = "1.0"` and stamp it into `node.metadata["schema_version"]` at the two LlamaIndex store task chokepoints. The field propagates automatically into PG `data_hierarchical_nodes.value->'__data__'->'metadata'->>'schema_version'` (via `PostgresDocumentStore`) and into Milvus' per-record `metadata` JSON (via `MilvusVectorStore`). No read-side changes in this repo for v1 — missing field is treated as implicit `"1.0"`. ## Boundaries & Constraints **Always:** - The string `"1.0"` MUST live in exactly one place — `airflow/dags/tasks/chunk_schema.py` — imported by both stamp sites. - Stamp only at the single-writer store-task chokepoints (`store_to_llamaindex_task`, `store_large_table_to_llamaindex_task`). Upstream chunk-builder tasks do NOT add the field — keeps v1 drift-proof. - Version format: semver string (`"."`). Start at `"1.0"`. - Existing tenant/node scalar-field stamping at `processing_tasks.py:2650-2653` stays exactly as-is; the schema-version stamp is an additional line in the same loop. **Ask First:** - If LlamaIndex `MilvusVectorStore` rejects the new metadata key (schema drift / strict mode), HALT. We expect LlamaIndex to serialize unknown keys into the Milvus `metadata` JSON field transparently. **Never:** - Do NOT declare a new Milvus scalar field in this ticket — scalar vs. JSON is a separate spike (deferred per GAP-043 analysis). - Do NOT backfill the ~existing PG/Milvus rows — the read-side contract is "missing = 1.0". - Do NOT modify `airflow/plugins/api/tags.py` or `domain_terms.py` readers — they already tolerate extra JSONB keys. - Do NOT publish the cross-service version contract in this repo — that spec goes through `textiq-blueprint` per `CLAUDE.md`. - Do NOT change chatbot-service or doc-generation — separate repos, separate tickets. ## I/O & Edge-Case Matrix | Scenario | Input / State | Expected Output / Behavior | Error Handling | |----------|---------------|---------------------------|----------------| | New hierarchical doc (text/pdf/pptx/xlsx detailed) | DAG runs → `store_to_llamaindex_task` fires | Every persisted node row has `value->'__data__'->'metadata'->>'schema_version' = '1.0'` in PG and `metadata.schema_version = "1.0"` in its Milvus record | N/A | | New large-table excel | DAG runs → `store_large_table_to_llamaindex_task` fires | Every persisted row carries `schema_version: "1.0"` in its metadata dict | N/A | | Legacy row (pre-v1.0) read by plugin | `tags.py` / `domain_terms.py` loads `value->'__data__'->'metadata'` with no `schema_version` key | Plugin behaves exactly as before; no version check added in v1 | Missing field is treated as implicit `"1.0"` by downstream contract | | Tag update on a legacy row | `PATCH /{chunk_id}/tags/status` runs `jsonb_set` on a row without `schema_version` | Update succeeds; row still has no `schema_version` (we do not retro-stamp) | N/A | ## Code Map - `airflow/dags/tasks/chunk_schema.py` — NEW. Exports `CHUNK_SCHEMA_VERSION = "1.0"` + module docstring describing the compatibility policy (missing = 1.0; consumers accept N and N-1; major = breaking, minor = additive). - `airflow/dags/tasks/processing_tasks.py:2650-2653` — extend the existing per-node injection loop in `store_to_llamaindex_task` to also set `node.metadata['schema_version'] = CHUNK_SCHEMA_VERSION`. - `airflow/dags/tasks/processing_tasks.py:2823-2839` — add `"schema_version": CHUNK_SCHEMA_VERSION` to the `Document(..., metadata={...})` dict inside `store_large_table_to_llamaindex_task`. - `airflow/dags/tasks/processing_tasks.py:2594-2623` — imports section inside `store_to_llamaindex_task`; add `from tasks.chunk_schema import CHUNK_SCHEMA_VERSION` (both store tasks use `@task.external_python` — imports must be local). - `airflow/dags/tasks/processing_tasks.py:2776-2788` — same import for `store_large_table_to_llamaindex_task`. - `tests/airflow/test_chunk_schema.py` — NEW. Asserts the constant is a valid semver-style string and matches the frozen `"1.0"` value so drift requires intentional test update. ## Tasks & Acceptance **Execution:** - [x] `airflow/dags/tasks/chunk_schema.py` -- create module with `CHUNK_SCHEMA_VERSION = "1.0"` plus a ~15-line docstring stating: the value, the bump rules (major = breaking, minor = additive), and the "missing = 1.0" consumer rule. Rationale: single source of truth, discoverable by future consumer repos. - [x] `airflow/dags/tasks/processing_tasks.py` (`store_to_llamaindex_task`) -- add `from tasks.chunk_schema import CHUNK_SCHEMA_VERSION` to the local imports block (~line 2624), then insert `node.metadata['schema_version'] = CHUNK_SCHEMA_VERSION` immediately after the tenant/node injection at line 2653. Rationale: hierarchical chunks (text/PDF/PPTX/Excel-detailed) stamp at the chokepoint. **Impl note:** added `sys.path.insert(0, '/opt/airflow/dags')` before the import because `/opt/airflow/dags` is not in the container `PYTHONPATH` (`docker/Dockerfile.airflow:82`), matching the existing pattern at `processing_tasks.py:1461`. - [x] `airflow/dags/tasks/processing_tasks.py` (`store_large_table_to_llamaindex_task`) -- add the same local import (~line 2787), then add `"schema_version": CHUNK_SCHEMA_VERSION,` to the `Document(..., metadata={...})` dict at ~line 2837. Rationale: large-table excel records stamp at their dedicated chokepoint. - [x] `tests/airflow/test_chunk_schema.py` -- assert `CHUNK_SCHEMA_VERSION == "1.0"` and that the string matches `^\d+\.\d+$`. Rationale: locks the constant; bumping the version requires an intentional test edit. **Acceptance Criteria:** - Given the code changes are applied and a hierarchical document is processed end-to-end, when I query `SELECT value->'__data__'->'metadata'->>'schema_version' FROM data_hierarchical_nodes WHERE document_id = :doc_id`, then every row returns `'1.0'`. - Given a large-table Excel document is processed, when I inspect its Milvus records' `metadata` JSON field, then every record contains `"schema_version": "1.0"`. - Given a legacy row exists without `schema_version`, when the tags plugin issues `PATCH /{chunk_id}/tags/status`, then the request succeeds and the row's other metadata keys are unchanged (no retro-stamp). - Given the constant module is imported from anywhere, when grep-searching the repo, then the string `"1.0"` for chunk schema appears only in `airflow/dags/tasks/chunk_schema.py` and `tests/airflow/test_chunk_schema.py`. ## Design Notes **Why stamp only at the two store tasks, not at every chunk builder.** DE has four upstream chunk builders (hierarchical text, hierarchical excel detailed, large-table excel, plus enrichment in `tag_records_nodes_task`). All flow through exactly two store tasks. Stamping at the chokepoint avoids N writers drifting out of sync when the version bumps. **Why `node.metadata["schema_version"]` and not a Milvus scalar field (yet).** LlamaIndex `MilvusVectorStore` serializes `node.metadata` into the collection's `metadata` JSON field without schema changes. That unblocks v1 with zero Milvus ops work. If consumers later need to filter/delete by version, a follow-up can promote it to a scalar field via `alter_collection_field` or a re-create migration. Noted in GAP-043 audit entry. **Why `"1.0"` (string, semver).** Plain string, sortable via `packaging.version`, extensible to `"1.1"` (additive) or `"2.0"` (breaking). Integer (`1`) would force a breaking bump for additive changes. ## Verification **Commands:** - `cd /home/nguyen/code/textiq-doc-extraction && uv run pytest tests/airflow/test_chunk_schema.py -v` -- expected: all pass. - `cd /home/nguyen/code/textiq-doc-extraction && uv run python -c "import sys; sys.path.insert(0, 'airflow/dags'); from tasks.chunk_schema import CHUNK_SCHEMA_VERSION; print(CHUNK_SCHEMA_VERSION)"` -- expected: prints `1.0`. **Manual checks:** - After a local DAG run on a sample doc, execute `SELECT value->'__data__'->'metadata'->>'schema_version' AS v, count(*) FROM data_hierarchical_nodes WHERE document_id = '' GROUP BY v;` — expect a single group with `v = '1.0'` and count matching the node count. - In Milvus, query a record from the collection and inspect its `metadata` field — expect `"schema_version": "1.0"` inside the JSON blob. ## Suggested Review Order **Single source of truth** - Canonical constant + bump rules + consumer contract; every other stop imports from here. [`chunk_schema.py:20`](../../airflow/dags/tasks/chunk_schema.py#L20) **Stamp chokepoints** - Hierarchical writer: stamp `schema_version` next to the existing tenant/node injection. [`processing_tasks.py:2654`](../../airflow/dags/tasks/processing_tasks.py#L2654) - Hierarchical writer: idempotent `sys.path` insert + local import (external-python constraint). [`processing_tasks.py:2628`](../../airflow/dags/tasks/processing_tasks.py#L2628) - Large-table writer: stamp `schema_version` inside the `Document(metadata={...})` literal. [`processing_tasks.py:2839`](../../airflow/dags/tasks/processing_tasks.py#L2839) - Large-table writer: matching idempotent import block. [`processing_tasks.py:2794`](../../airflow/dags/tasks/processing_tasks.py#L2794) **Drift lock** - Lock test pins the constant to `"1.0"` and enforces `.` format. [`test_chunk_schema.py:17`](../../tests/airflow/test_chunk_schema.py#L17)