docs: structured data ingestion design proposal

docs/ingestion.md

@@ -4,6 +4,11 @@ When you call `rag.ingest("document.txt")`, the SDK transforms your raw text int
 
 This document explains what each step does, why it exists, and how to tune it.
 
+!!! info "Structured sources (CSV, JSON, tables, existing graphs)"
+    This page describes the **unstructured** path — prose in, LLM extraction, graph out.
+    Structured inputs skip LLM extraction entirely and are covered by a separate proposal:
+    [Design: Structured Data Ingestion](design/structured-ingestion.md).
+
 ---
 
 ## The Big Picture

mkdocs.yml

@@ -54,3 +54,5 @@ nav:
   - Ontology Evolution: ontology-evolution.md
   - Benchmark: benchmark.md
   - API Reference: api-reference.md
+  - Design Proposals:
+      - Structured Data Ingestion: design/structured-ingestion.md

poc/structured-ingestion/FINDINGS.md

@@ -0,0 +1,125 @@
+# FINDINGS — what the spikes changed about the design
+
+Five spikes, each answering one open question against real code and a real FalkorDB.
+`python run_all.py` → **all 5 pass**.
+
+Four of the five falsified something in `docs/design/structured-ingestion.md`. One of those
+(s3) inverts a headline decision.
+
+| # | Spike | Verdict on the design |
+| --- | --- | --- |
+| s1 | record stream | **amend** — `RecordBatch` must expose a stream *factory*; as written it silently writes zero rows |
+| s2 | mapping DSL | **amend** — edges must address node *aliases*, not labels; `to_ontology()` needs two guards |
+| s3 | identity | **invert** — `identity=["name"]` as the default is wrong; key + `alias_ids` is the only policy that works |
+| s4 | record-as-chunk | **confirmed, incl. the predicted data-loss trap** |
+| s5 | pipeline seam | **amend** — share a base class, don't subclass; one new kwarg for `NEXT_CHUNK` |
+
+---
+
+## 1. `RecordBatch` must be a stream factory  → proposal #1
+
+Pydantic v2 keeps an `Iterable[dict]` field lazy (200k rows: ~0 MB vs 71.6 MB materialised) but
+wraps it in a one-shot `ValidatorIterator`. Proposal #6 iterates records **twice** — step 3 builds
+record chunks, step 4 maps records to `GraphData`. Measured result:
+
+```
+step 3 saw 10 records, step 4 saw 0 — no error raised
+```
+
+Silent zero-row ingest. Also, the annotation erases list-ness: `len()` raises even when the caller
+passed a list, so no cheap record count is available for progress or `IngestionResult`.
+
+**Change:** `open_records: Callable[[], Iterator[dict]]` + `record_count: int | None`. Verified
+re-iterable and `model_dump()`-safe. *(details: `s1_record_stream/NOTES.md`)*
+
+## 2. Edges must address aliases, not labels  → proposal #2
+
+`EdgeMapping(source="Organization", target="Organization")` cannot express a transaction with a
+buyer and a seller. Executed against `transactions.csv` it produced a **self-loop**
+(`ORG-7 -> ORG-7` instead of `ORG-7 -> ORG-42`) with no error. Buyer/seller, manager/report,
+parent/subsidiary is the standard shape of transactional data, not an edge case.
+
+**Change:** `NodeMapping(alias=..., ...)`, `EdgeMapping(source=<alias>, target=<alias>)`, alias
+defaulting to the label so the 80% case is untouched. Two further guards on `to_ontology()`: reject
+`_RESERVED_ATTRIBUTE_NAMES` (a mapping can otherwise shadow SDK-written keys like `id` and
+`description`), and emit stubs for reference-only labels (otherwise `Ontology`'s own validator
+warns on every ingest — 2 warnings from one mapping in the spike).
+
+Bonus: `Entity(label=..., identity=[...])` **already works** via `Config.extra = "allow"` and
+survives `model_dump()`, so #3 is prototypable with zero `src` changes and old `ontology.json`
+files stay loadable. *(details: `s2_mapping_dsl/NOTES.md`)*
+
+## 3. Identity: the design's default is backwards  → proposals #3, #4
+
+Measured on the #82 acceptance corpus, three policies × two ingest orders:
+
+| Policy | Acme nodes | #82 traversal |
+| --- | --- | --- |
+| name-first (`identity=["name"]`, **the design's default**) | 2 | **0 people reachable** |
+| key-only | 2 | **0 people reachable** |
+| key + `alias_ids` + resolve pass | **1** | **2 people reachable** |
+
+Cause: `employees.csv` references its org by `org_id=ORG-42` and has no `org_name` column. Under
+name-first identity the mapping **cannot compute the identity of the entity it points at**, so
+proposal #2's rule "each mapping must supply the type's identity attributes" is unsatisfiable for
+any normalised foreign key. The result is a stub node holding all the `WORKS_AT` edges, sitting
+next to the real Acme node that holds the prose — the exact failure #82 exists to prevent, silent.
+
+**Change:** structured writes are **key-identified**, and `alias_ids` moves from optional bridge to
+**critical path**. Caveat, also measured: with no source carrying key *and* name together, the
+bridge degrades to 2 nodes / 0 reachable. The honest contract is **ingest order does not matter,
+but presence does** — and unbridged stubs must be reported in `IngestionResult` rather than left
+silent. *(details: `s3_identity/NOTES.md`)*
+
+## 4. The predicted cutover trap is real  → proposal #5
+
+The design review predicted, from reading code, that deterministic chunk uids keyed on the
+canonical document id would destroy data during `update()`. Run against real FalkorDB through the
+real `GraphStore`:
+
+| chunk uid keyed on | chunks before | shared with pending | **after cutover** |
+| --- | --- | --- | --- |
+| canonical doc id | 3 | 3 | **0** |
+| effective (pending) doc id | 3 | 0 | **3** |
+
+The pending run `MERGE`s onto the live document's chunk nodes; `rollforward_cutover()` step 1
+deletes them; step 3 promotes an empty document. No exception. Today's `uuid4()` uids are
+accidentally immune — which is why making them deterministic is the dangerous part of #5.
+
+Everything else in #5 **holds**: `get_document_entity_candidates()` found 5 record-chunk entities,
+`delete_stale_relationships()` GC'd exactly the removed row's fact, `delete_orphan_entities()`
+removed exactly Carol and the org that lost its last mention.
+
+**Change:** promote the §5 callout from "review finding" to a hard rule with this measurement
+attached. §8.8 (row-level incremental update is *not* free) stays refuted.
+*(details: `s4_record_as_chunk/NOTES.md`)*
+
+## 5. Share a base class, don't subclass  → proposal #6
+
+`_build_lexical_graph` / `_prune` / `_write_mentions` are reusable **verbatim** — the first already
+consumes `TextChunks`, which is what proposal #5 turns records into. Both factorings produced
+identical graphs (9 nodes, 6 `MENTIONED_IN`).
+
+But `IngestionPipeline.__init__` requires `loader, chunker, extractor, resolver, graph_store,
+vector_store`. A structured pipeline has no chunker and no LLM extractor, so subclassing means
+passing `None` and hoping — it works today only by accident of which methods are called.
+
+**Change:** extract the three methods into a `LexicalGraphWriter` base depending only on
+`graph_store`. Plus one new kwarg: `_build_lexical_graph(..., link_sequential: bool = True)`, since
+reusing it as-is chains `NEXT_CHUNK` between unrelated CSV rows (N-1 edges asserting a sequence
+that doesn't exist, while `cypher_generation.py` tells the LLM those edges mean "next sequential
+Chunk"). *(details: `s5_pipeline_seam/NOTES.md`)*
+
+---
+
+## What did **not** change
+
+The core architecture survived contact with the database. Record-as-chunk, `RELATES` + `rel_type`,
+mapping-as-ontology-fragment, deterministic no-LLM mapping, and "no retrieval strategy is touched"
+all held up. Every correction above is a change to a *signature or a default*, not to the shape of
+the design — which is the outcome a spike round is supposed to produce.
+
+## Disposal
+
+Once these are folded into `docs/design/structured-ingestion.md` and the implementation lands,
+delete `poc/`. Nothing imports it; nothing ships it.

poc/structured-ingestion/README.md

@@ -0,0 +1,42 @@
+# Structured-ingestion spikes
+
+Throwaway experiments backing [`docs/design/structured-ingestion.md`](../../docs/design/structured-ingestion.md)
+(tracking: FalkorDB/research#82).
+
+**This folder is disposable.** Nothing here ships. It is outside the wheel
+(`[tool.hatch.build.targets.wheel] packages = ["src/graphrag_sdk"]`), outside pytest
+(`testpaths = ["tests"]`), and outside CI (every job sets `working-directory: graphrag_sdk`
+and lints only `src/`). Delete the whole directory once the findings are folded into the design
+and the real implementation lands.
+
+## Why it exists
+
+The design doc makes claims. These spikes exist to **find out where the claims are wrong** before
+we build on them — each folder answers *one specific open question* and records a decision, rather
+than re-implementing a proposal. Five spikes, not twelve; a proposal without a genuine open
+question does not get a folder.
+
+| Spike | Proposal | The question it answers |
+| --- | --- | --- |
+| `s1_record_stream` | #1 | Can `RecordBatch` actually hold a lazy record stream, or does the model layer consume it? What does streaming buy in memory? |
+| `s2_mapping_dsl` | #2 | Which DSL shape expresses all four real-world record shapes without special cases — and does `to_ontology()` round-trip into a valid `Ontology`? |
+| `s3_identity` | #3, #4 | Do the three candidate identity policies actually produce one connected graph? Measured, not argued. |
+| `s4_record_as_chunk` | #5 | Does record-as-chunk really inherit orphan cleanup unchanged — and is the predicted `update()` cutover data-loss trap real? |
+| `s5_pipeline_seam` | #6 | Can `StructuredIngestionPipeline` reuse the existing pipeline's steps as-is, or do the signatures need to change? |
+
+## Running
+
+```bash
+cd poc/structured-ingestion
+python run_all.py             # everything; DB spikes skip if FalkorDB is unreachable
+python s3_identity/spike.py   # one spike
+```
+
+Spikes needing a graph use `FALKOR_HOST` / `FALKOR_PORT` (default `localhost:6379`) and write to
+throwaway graph names prefixed `poc_`. `docker compose up -d falkordb` from the repo root starts one.
+
+No API keys are needed anywhere — the harness supplies a deterministic fake embedder, and the
+spikes never call an LLM (which is the point of the whole design).
+
+Each spike prints its findings and writes them to its own `NOTES.md`. [`FINDINGS.md`](FINDINGS.md)
+rolls up the decisions that feed back into the design doc.

poc/structured-ingestion/_harness/env.py

@@ -0,0 +1,119 @@
+"""Shared harness for the structured-ingestion spikes.
+
+Importing this module pins *this worktree's* ``src`` ahead of any installed
+graphrag_sdk, so spikes exercise the branch under review rather than whatever
+``pip install -e`` happens to point at.
+"""
+
+from __future__ import annotations
+
+import os
+import sys
+from pathlib import Path
+
+# poc/structured-ingestion/_harness/env.py -> repo root
+REPO_ROOT = Path(__file__).resolve().parents[3]
+SDK_SRC = REPO_ROOT / "graphrag_sdk" / "src"
+FIXTURES = Path(__file__).resolve().parent / "fixtures"
+
+if str(SDK_SRC) not in sys.path:
+    sys.path.insert(0, str(SDK_SRC))
+
+FALKOR_HOST = os.getenv("FALKOR_HOST", "localhost")
+FALKOR_PORT = int(os.getenv("FALKOR_PORT", "6379"))
+
+
+def sdk_is_local() -> bool:
+    """True when ``graphrag_sdk`` resolves inside this worktree."""
+    import graphrag_sdk
+
+    return str(SDK_SRC) in str(Path(graphrag_sdk.__file__).resolve())
+
+
+def falkor_available() -> bool:
+    """Cheap reachability probe so DB spikes can skip instead of exploding."""
+    import socket
+
+    try:
+        with socket.create_connection((FALKOR_HOST, FALKOR_PORT), timeout=2):
+            return True
+    except OSError:
+        return False
+
+
+def connection(graph_name: str):
+    """A FalkorDBConnection against a throwaway ``poc_``-prefixed graph."""
+    from graphrag_sdk.core.connection import ConnectionConfig, FalkorDBConnection
+
+    if not graph_name.startswith("poc_"):
+        raise ValueError(f"refusing non-throwaway graph name {graph_name!r}; use a poc_ prefix")
+    return FalkorDBConnection(
+        ConnectionConfig(host=FALKOR_HOST, port=FALKOR_PORT, graph_name=graph_name)
+    )
+
+
+async def reset_graph(conn) -> None:
+    """Drop everything in the connected graph. Only ever called on poc_ graphs."""
+    await conn.query("MATCH (n) DETACH DELETE n")
+
+
+class FakeEmbedder:
+    """Deterministic, dependency-free embedder.
+
+    Spikes care about *what gets written and traversed*, never about vector
+    quality, so a hash-derived vector is sufficient and keeps the whole folder
+    runnable with no API keys.
+    """
+
+    def __init__(self, dimension: int = 8) -> None:
+        self.dimension = dimension
+        self.calls = 0
+
+    @property
+    def model_name(self) -> str:
+        return "fake-embedder"
+
+    def _vec(self, text: str) -> list[float]:
+        import hashlib
+
+        digest = hashlib.sha256(text.encode("utf-8")).digest()
+        return [digest[i] / 255.0 for i in range(self.dimension)]
+
+    async def aembed_query(self, text: str) -> list[float]:
+        self.calls += 1
+        return self._vec(text)
+
+    async def aembed_documents(self, texts: list[str]) -> list[list[float]]:
+        self.calls += len(texts)
+        return [self._vec(t) for t in texts]
+
+
+# ── tiny reporting helpers ───────────────────────────────────────
+
+
+class Report:
+    """Collects PASS/FAIL checks so a spike ends with a verdict, not a wall of prints."""
+
+    def __init__(self, title: str) -> None:
+        self.title = title
+        self.lines: list[str] = []
+        self.failures = 0
+        print(f"\n=== {title} ===")
+
+    def check(self, ok: bool, label: str, detail: str = "") -> bool:
+        mark = "PASS" if ok else "FAIL"
+        if not ok:
+            self.failures += 1
+        line = f"[{mark}] {label}" + (f" — {detail}" if detail else "")
+        self.lines.append(line)
+        print(line)
+        return ok
+
+    def note(self, text: str) -> None:
+        self.lines.append(f"       {text}")
+        print(f"       {text}")
+
+    def verdict(self) -> int:
+        status = "OK" if self.failures == 0 else f"{self.failures} FAILED"
+        print(f"--- {self.title}: {status}")
+        return 1 if self.failures else 0

poc/structured-ingestion/_harness/fixtures/acme_report.txt

@@ -0,0 +1,5 @@
+Acme Corp reported a Q3 revenue miss, attributing the shortfall to delayed
+enterprise renewals. Globex, its largest counterparty, expanded its own
+services division over the same period.
+
+Alice Smith, an engineer at Acme Corp, presented the remediation plan.

poc/structured-ingestion/_harness/fixtures/catalog.json

@@ -0,0 +1,44 @@
+{
+  "catalog": {
+    "version": 3
+  },
+  "products": [
+    {
+      "sku": "SKU-1",
+      "name": "Anvil",
+      "price": 99.5,
+      "sold_by": {
+        "org_id": "ORG-42",
+        "org_name": "Acme Corp"
+      },
+      "tags": [
+        "hardware",
+        "heavy"
+      ]
+    },
+    {
+      "sku": "SKU-2",
+      "name": "Rocket Skates",
+      "price": 249.0,
+      "sold_by": {
+        "org_id": "ORG-42",
+        "org_name": "Acme Corp"
+      },
+      "tags": [
+        "hardware"
+      ]
+    },
+    {
+      "sku": "SKU-9",
+      "name": "Consulting Hour",
+      "price": 180.0,
+      "sold_by": {
+        "org_id": "ORG-7",
+        "org_name": "Globex"
+      },
+      "tags": [
+        "services"
+      ]
+    }
+  ]
+}

poc/structured-ingestion/_harness/fixtures/employees.csv

@@ -0,0 +1,4 @@
+employee_id,full_name,age,job_title,org_id,start_date
+E-1,Alice Smith,34,Engineer,ORG-42,2019-04-01
+E-2,Bob Jones,45,CFO,ORG-42,2015-11-15
+E-3,Carol White,29,Engineer,ORG-7,2021-06-30

poc/structured-ingestion/_harness/fixtures/orgs.csv

@@ -0,0 +1,3 @@
+org_id,org_name,hq_country,employee_count
+ORG-42,Acme Corp,US,1200
+ORG-7,Globex,GB,340

poc/structured-ingestion/_harness/fixtures/transactions.csv

@@ -0,0 +1,3 @@
+txn_id,buyer_org_id,seller_org_id,amount,currency,closed_on
+TXN-100,ORG-7,ORG-42,250000,USD,2024-02-11
+TXN-101,ORG-42,ORG-7,90000,GBP,2024-05-02