---
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
---

<frozen-after-approval reason="human-owned intent — do not modify unless human renegotiates">

## 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 (`"<major>.<minor>"`). 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 |

</frozen-after-approval>

## 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 = '<uuid>' 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 `<major>.<minor>` format.
  [`test_chunk_schema.py:17`](../../tests/airflow/test_chunk_schema.py#L17)
