feat(graph): bulk_upsert_entity_with_lineage_parts Protocol + 4 backends + curation cutover (W8-2) (#1771)

Wave 8 task #13 (W8-2): adds a bulk-upsert primitive on the
``LineageGraphStore`` Protocol so callers consolidating N×M
``(description_part, lineage_member)`` tuples to the same target
entity get one transaction / one round-trip instead of N×M
sequential single upserts. Cuts ``LineageEntityMerger.merge_entities``
step 6a over to the bulk path.

## Why

``GraphCurationService.merge_entities`` (the N-source-into-1-target
user merge flow) re-anchors every source's description parts under
the target name. With N source entities each carrying M description
parts, that loop emitted N×M sequential ``upsert_entity_with_lineage``
round-trips — one full transaction per part. For typical curation
runs (3-10 sources × 5-20 parts each), that's 15-200 sequential SQL
hits where one bulk write would do.

Per architect spec (Wave 8 candidate W8-2 sediment) + huangheng's
task #6 PR #1758 perf observation, this folds into a single bulk
write per backend.

## Scope (5 numbered items)

1. **Protocol method** — ``bulk_upsert_entity_with_lineage_parts(*,
   parts: Sequence[tuple[EntityRecord, LineageMember]])`` on
   ``aperag/indexing/graph.py:LineageGraphStore``. All ``record.name``
   values MUST share the same string (asserted as ``ValueError``).
   Empty parts is a no-op. Per-part dedup key is
   ``(document_id, parse_version)`` last-wins. Bulk path NEVER
   touches ``compacted_description`` (preserves existing — same
   ``COALESCE``-style semantic as single upsert with ``None``).
2. **InMemory ref impl** — single ``asyncio.Lock``-guarded loop in
   ``InMemoryLineageGraphStore``.
3. **Postgres impl** — single ``INSERT … ON CONFLICT (collection_id,
   name) DO UPDATE`` that strips matching keys via
   ``jsonb_array_elements`` ``NOT EXISTS`` against an incoming
   ``strip_keys_json`` array, then appends the whole new
   ``new_members_json`` / ``new_parts_json`` arrays. One statement,
   atomic.
4. **Neo4j impl** — single Cypher MERGE + parallel-list strip-then-
   append, with the strip predicate matching against the **set** of
   incoming keys (``IN $strip_keys`` on the ``"<doc_id>|<pv>"`` key
   string). Row-lock on MERGE serialises concurrent bulk ops on the
   same entity.
5. **Nebula impl** — single ``EntityLock(target_name)`` acquire +
   single read / Python merge / write. Mirrors the existing
   read-modify-write pattern of single upsert but folds the strip-
   then-append over the **set** of incoming keys.

## Caller cutover

* ``LineageEntityMerger.merge_entities`` (step 6a) — replaces the
  N×M ``for src in source_entities: for part in
  src.description_parts: await self._store.upsert_entity_with_lineage(...)``
  with a single ``bulk_upsert_entity_with_lineage_parts`` call.
  Step 6b's sentinel write (``__curation_merge__`` final write with
  unified+compacted text) still goes through the single-upsert path
  because it needs the ``compacted_description`` column write that
  the bulk path intentionally doesn't carry.

## Alias-redirect decorator

* ``LineageGraphStoreWithAliasRedirect`` — bulk path mirrors single
  upsert: each part's ``record.name`` resolves through the alias
  map before forwarding to the inner store. The merger always
  passes records pinned to the canonical ``final_target`` so the
  redirect is a no-op in that flow, but symmetry with the single
  upsert contract means a future caller writing to an aliased name
  still gets correct behaviour.

## 12-invariant cross-check (Wave 7 §K.12)

* **#1 L1 不污染**: bulk path operates on lineage SETs only — kg.jsonl
  raw extract layer untouched. ✅
* **#3 indexer write redirect through alias map**: bulk path goes
  through the same decorator alias resolution as single upsert. ✅
* **#9 alias redirect transparent**: decorator forwards bulk to
  inner with redirected names; merger callers see no behaviour
  change. ✅
* **#10 DB column cap**: Postgres ``compacted_description`` Text
  column unchanged (bulk path preserves existing value via
  COALESCE-equivalent — bulk passes ``None`` end-to-end through
  the Postgres SQL → INSERT branch sets NULL, ON CONFLICT branch
  preserves). ✅
* All other invariants: unaffected.

## 4-pattern pre-check matrix

* Pattern A (kg.jsonl shape): unchanged
* Pattern B (Lineage SET semantics): bulk preserves dedup-by-
  ``(document_id, parse_version)`` exactly — same key as single
  upsert. ✅
* Pattern C (Cypher LIST<MAP>): bulk reuses parallel-list encoding
  for the strip-then-append. ✅
* Pattern D (vector 3-field payload): unchanged

## Simple-stable 4-guardrail

* **#1 不无限扩范围**: 1 new Protocol method, no new public REST
  surface, no new schema column, no new alembic migration.
* **#2 尽快上线**: single PR, no spec amend needed (architect
  W8-2 candidate sediment + Wave 7 task #1 3-backend Protocol
  pattern reuse).
* **#3 简单稳定**: bulk semantic mirrors single upsert exactly;
  caller cutover is a 1-call replacement of the N×M loop.
* **#4 私有化部署免维护**: backend-portable (4 backends shipped
  cross-backend); no operator-facing config knob.

## Test plan

- [x] InMemory contract — 7 new tests pin empty / create / replace /
      mismatched-names / dedup-within-input / preserves-compacted /
      last-entity-type-wins.
- [x] Alias-redirect decorator — 3 new tests pin redirect-each-part /
      no-alias-passthrough / empty-short-circuit.
- [x] LineageEntityMerger step 6a cutover — existing
      ``test_source_parts_reanchored_preserving_doc_lineage``
      rewritten to assert the new single bulk call (was: 3
      sequential single upserts) + single sentinel final write.
- [x] All 1166 unit tests pass (up from 1141 baseline; 25 new tests).
- [x] Wave 7 grep-zero contract — 10/10 still pass (intent-driven
      gate unaffected).
- [x] ruff format / check clean on touched files.
- [ ] Cross-backend integration test — sediment to Wave 8 follow-up
      (out of scope this PR; the 4-backend Protocol contract is
      structurally identical to single upsert which already has
      cross-backend integration coverage).
- [ ] CR by @huangheng (focus: invariant #1 L1, #9 alias redirect
      transparent, 4-backend cross-roundtrip on contract level).

Author: earayu

aperag/graph_curation/lineage_merge.py

@@ -260,28 +260,35 @@ async def merge_entities(
 
         # Step 6a — re-anchor source parts under the target name,
         # preserving their original lineage so per-doc tracking is not
-        # lost (invariant #1). DescriptionPart carries no chunk_ids of
-        # its own; we look up the matching LineageMember by
-        # ``(document_id, parse_version)`` to recover them.
+        # lost (invariant #1). Wave 8 W8-2: the N×M loop is folded
+        # into a single ``bulk_upsert_entity_with_lineage_parts`` call
+        # so the whole consolidation is one transaction / one round-
+        # trip per backend, not N×M sequential upserts.
+        bulk_parts: list[tuple[EntityRecord, LineageMember]] = []
         for src in source_entities:
             lineage_by_key = {m.key(): m for m in src.source_lineage}
+            tenant = self._tenant_scope_key_for(src)
             for part in src.description_parts:
                 member = lineage_by_key.get(part.key())
                 chunk_ids = tuple(member.chunk_ids) if member is not None else ()
-                await self._store.upsert_entity_with_lineage(
-                    record=EntityRecord(
-                        name=final_target,
-                        entity_type=target_entity.entity_type,
-                        description=part.text,
-                        source_chunk_ids=chunk_ids,
-                    ),
-                    lineage=LineageMember(
-                        document_id=part.document_id,
-                        parse_version=part.parse_version,
-                        tenant_scope_key=self._tenant_scope_key_for(src),
-                        chunk_ids=chunk_ids,
-                    ),
+                bulk_parts.append(
+                    (
+                        EntityRecord(
+                            name=final_target,
+                            entity_type=target_entity.entity_type,
+                            description=part.text,
+                            source_chunk_ids=chunk_ids,
+                        ),
+                        LineageMember(
+                            document_id=part.document_id,
+                            parse_version=part.parse_version,
+                            tenant_scope_key=tenant,
+                            chunk_ids=chunk_ids,
+                        ),
+                    )
                 )
+        if bulk_parts:
+            await self._store.bulk_upsert_entity_with_lineage_parts(parts=bulk_parts)
 
         # Step 6b — final write with unified text + compacted_description
         # under the curation-merge sentinel lineage.

aperag/indexing/alias_redirect_store.py

@@ -163,6 +163,34 @@ async def upsert_relation_with_lineage(
             compacted_description=compacted_description,
         )
 
+    async def bulk_upsert_entity_with_lineage_parts(self, *, parts) -> None:
+        # Wave 8 W8-2: alias-redirect on the bulk write surface,
+        # mirror of the single-upsert path. Resolve each ``record.name``
+        # through the alias map; if any tuple's name redirects, the
+        # whole bulk write retargets to the canonical name. The
+        # ``LineageEntityMerger`` is the only known caller and always
+        # passes records already pinned to the canonical
+        # ``final_target`` so the redirect is a no-op in that flow —
+        # but we apply it here for symmetry with the single upsert
+        # contract (a future caller writing to an aliased name still
+        # gets the right behaviour).
+        if not parts:
+            return
+        redirected: list[tuple[EntityRecord, LineageMember]] = []
+        for record, lineage in parts:
+            canonical = await self._alias_repo.resolve_canonical(collection_id=self._collection_id, name=record.name)
+            if canonical != record.name:
+                logger.debug(
+                    "alias_redirect: bulk entity part %r → %r (collection=%s)",
+                    record.name,
+                    canonical,
+                    self._collection_id,
+                )
+                redirected.append((replace(record, name=canonical), lineage))
+            else:
+                redirected.append((record, lineage))
+        await self._inner.bulk_upsert_entity_with_lineage_parts(parts=redirected)
+
     # ------------------------------------------------------------------
     # Passthrough — forward every non-redirected Protocol method
     # unchanged. Pinned by

aperag/indexing/graph.py

@@ -572,6 +572,39 @@ async def upsert_relation_with_lineage(
         a non-None string overwrites.
         """
 
+    async def bulk_upsert_entity_with_lineage_parts(
+        self,
+        *,
+        parts: Sequence[tuple[EntityRecord, LineageMember]],
+    ) -> None:
+        """Wave 8 W8-2: atomic bulk variant of
+        :meth:`upsert_entity_with_lineage`.
+
+        Each ``(record, lineage)`` tuple lands as a separate description
+        part with the same ``(document_id, parse_version)`` dedup key
+        the single upsert uses; semantically equivalent to looping
+        :meth:`upsert_entity_with_lineage` but executed in **one
+        transaction / one round-trip** so callers consolidating N×M
+        parts (e.g. :class:`LineageEntityMerger.merge_entities` step 6a)
+        get O(1) network round-trips instead of O(N×M).
+
+        Contract:
+
+        * All ``record.name`` values MUST share the same string —
+          backends MAY assert and raise ``ValueError`` if they don't.
+        * Empty ``parts`` is a no-op.
+        * Per-part ``record.entity_type`` follows the single-upsert
+          "most recently observed value wins" rule (last tuple's type
+          is the post-write entity_type for the row).
+        * ``compacted_description`` is intentionally **not** a
+          parameter — the bulk path never touches the column. Callers
+          that need to set it run a separate single
+          :meth:`upsert_entity_with_lineage` afterwards (the merger's
+          step 6b sentinel write does exactly this).
+        * Forward-only retry safety: the dedup key is per-part, so a
+          mid-flight crash + retry replays each tuple idempotently.
+        """
+
     async def get_entity(self, entity_name: str) -> EntityWithLineage | None:
         """Read-path helper used by tests / read primitives. Returns
         the canonical lineage view, or ``None`` if the row was GC'd.
@@ -893,6 +926,32 @@ async def upsert_relation_with_lineage(
             if compacted_description is not None:
                 row.compacted_description = compacted_description
 
+    async def bulk_upsert_entity_with_lineage_parts(
+        self,
+        *,
+        parts: Sequence[tuple[EntityRecord, LineageMember]],
+    ) -> None:
+        if not parts:
+            return
+        target_name = parts[0][0].name
+        if any(record.name != target_name for record, _ in parts):
+            raise ValueError("bulk_upsert_entity_with_lineage_parts: all records must share the same name")
+        async with self._guard:
+            row = self._entities.get(target_name)
+            if row is None:
+                row = _InMemoryEntityRow(name=target_name, entity_type=parts[0][0].entity_type)
+                self._entities[target_name] = row
+            for record, lineage in parts:
+                # Type may evolve as new docs refine the entity; keep
+                # the most recently observed value (mirror single upsert).
+                row.entity_type = record.entity_type
+                row.source_lineage[lineage.key()] = lineage
+                row.description_parts[lineage.key()] = DescriptionPart(
+                    document_id=lineage.document_id,
+                    parse_version=lineage.parse_version,
+                    text=record.description,
+                )
+
     # ---- read path --------------------------------------------------
 
     async def get_entity(self, entity_name: str) -> EntityWithLineage | None:

aperag/indexing/graph_storage/nebula.py

@@ -845,6 +845,72 @@ def _upsert() -> None:
 
             await asyncio.to_thread(_upsert)
 
+    async def bulk_upsert_entity_with_lineage_parts(
+        self,
+        *,
+        parts,
+    ) -> None:
+        """Wave 8 W8-2: Nebula bulk variant — single ``EntityLock``
+        acquire + single read-modify-write applies the whole ``parts``
+        list. Reuses the read/Python-merge/write pattern of
+        :meth:`upsert_entity_with_lineage` but folds the strip-then-
+        append over the **set** of incoming keys, so N×M parts collapse
+        to one write.
+        """
+        if not parts:
+            return
+        target_name = parts[0][0].name
+        if any(record.name != target_name for record, _ in parts):
+            raise ValueError("bulk_upsert_entity_with_lineage_parts: all records must share the same name")
+
+        # Dedup last-wins by (document_id, parse_version).
+        deduped: dict[tuple[str, str], tuple[EntityRecord, LineageMember]] = {}
+        for record, lineage in parts:
+            deduped[(lineage.document_id, lineage.parse_version)] = (record, lineage)
+
+        new_members_in: list[LineageMember] = []
+        new_parts_in: list[DescriptionPart] = []
+        last_entity_type: str = parts[0][0].entity_type
+        for record, lineage in deduped.values():
+            new_members_in.append(lineage)
+            new_parts_in.append(
+                DescriptionPart(
+                    document_id=lineage.document_id,
+                    parse_version=lineage.parse_version,
+                    text=record.description,
+                )
+            )
+            last_entity_type = record.entity_type
+
+        keys_to_strip = set(deduped.keys())
+
+        await self.ensure_schema()
+        async with self._entity_lock.acquire(target_name):
+
+            def _upsert() -> None:
+                row = self._read_entity_lineage(target_name)
+                if row is None:
+                    merged_members = list(new_members_in)
+                    merged_parts = list(new_parts_in)
+                    existing_compacted: str | None = None
+                else:
+                    _existing_type, members, parts_existing, existing_compacted = row
+                    kept_members = [m for m in members if m.key() not in keys_to_strip]
+                    kept_parts = [p for p in parts_existing if p.key() not in keys_to_strip]
+                    merged_members = kept_members + new_members_in
+                    merged_parts = kept_parts + new_parts_in
+                self._write_entity_vertex(
+                    name=target_name,
+                    type_value=last_entity_type,
+                    source_lineage=merged_members,
+                    description_parts=merged_parts,
+                    # Bulk path never touches compacted_description
+                    # (preserves existing, mirror Postgres / Neo4j).
+                    compacted_description=existing_compacted,
+                )
+
+            await asyncio.to_thread(_upsert)
+
     async def upsert_relation_with_lineage(
         self,
         *,

aperag/indexing/graph_storage/neo4j.py

@@ -440,6 +440,98 @@ async def upsert_entity_with_lineage(
                 compacted_description=compacted_description,
             )
 
+    async def bulk_upsert_entity_with_lineage_parts(
+        self,
+        *,
+        parts,
+    ) -> None:
+        """Wave 8 W8-2: bulk variant — single Cypher statement covers
+        the whole ``parts`` list. Strip-then-append is expressed against
+        the **set** of incoming ``(document_id, parse_version)`` keys
+        rather than a single key, so the MERGE row-lock still serialises
+        concurrent bulk ops against the same entity.
+        """
+        if not parts:
+            return
+        target_name = parts[0][0].name
+        if any(record.name != target_name for record, _ in parts):
+            raise ValueError("bulk_upsert_entity_with_lineage_parts: all records must share the same name")
+
+        # Dedup last-wins by (document_id, parse_version).
+        deduped: dict[tuple[str, str], tuple[EntityRecord, LineageMember]] = {}
+        for record, lineage in parts:
+            deduped[(lineage.document_id, lineage.parse_version)] = (record, lineage)
+
+        new_member_jsons: list[str] = []
+        new_part_jsons: list[str] = []
+        new_doc_ids: list[str] = []
+        new_parse_versions: list[str] = []
+        last_entity_type: str = parts[0][0].entity_type
+        for record, lineage in deduped.values():
+            new_member_jsons.append(_lineage_member_json(lineage))
+            new_part_jsons.append(
+                _description_part_json(
+                    DescriptionPart(
+                        document_id=lineage.document_id,
+                        parse_version=lineage.parse_version,
+                        text=record.description,
+                    )
+                )
+            )
+            new_doc_ids.append(lineage.document_id)
+            new_parse_versions.append(lineage.parse_version)
+            last_entity_type = record.entity_type
+
+        # ``strip_keys`` is a list of strings ``"<doc_id>|<parse_version>"`` —
+        # Cypher list-of-string membership is the easiest way to express
+        # "drop element if its key matches *any* incoming key" without
+        # nested list-comprehensions over a list of maps (which is harder
+        # to read and the same complexity).
+        strip_keys = [f"{doc_id}|{parse_version}" for (doc_id, parse_version) in deduped.keys()]
+
+        query = (
+            f"MERGE (n:{_ENTITY_LABEL} {{collection_id: $collection_id, name: $name}}) "
+            f"ON CREATE SET "
+            f"  n.source_lineage = [], "
+            f"  n.source_lineage_doc_ids = [], "
+            f"  n.source_lineage_parse_versions = [], "
+            f"  n.description_parts = [], "
+            f"  n.description_parts_doc_ids = [], "
+            f"  n.description_parts_parse_versions = [], "
+            f"  n.compacted_description = NULL, "
+            f"  n.gmt_created = datetime() "
+            f"WITH n, "
+            f"  [i IN range(0, size(n.source_lineage_doc_ids) - 1) "
+            f"   WHERE NOT (n.source_lineage_doc_ids[i] + '|' + n.source_lineage_parse_versions[i]) "
+            f"             IN $strip_keys] AS sl_keep, "
+            f"  [i IN range(0, size(n.description_parts_doc_ids) - 1) "
+            f"   WHERE NOT (n.description_parts_doc_ids[i] + '|' + n.description_parts_parse_versions[i]) "
+            f"             IN $strip_keys] AS dp_keep "
+            f"SET n.entity_type = $entity_type, "
+            f"    n.source_lineage = [i IN sl_keep | n.source_lineage[i]] + $new_member_jsons, "
+            f"    n.source_lineage_doc_ids = [i IN sl_keep | n.source_lineage_doc_ids[i]] + $new_doc_ids, "
+            f"    n.source_lineage_parse_versions = "
+            f"      [i IN sl_keep | n.source_lineage_parse_versions[i]] + $new_parse_versions, "
+            f"    n.description_parts = [i IN dp_keep | n.description_parts[i]] + $new_part_jsons, "
+            f"    n.description_parts_doc_ids = "
+            f"      [i IN dp_keep | n.description_parts_doc_ids[i]] + $new_doc_ids, "
+            f"    n.description_parts_parse_versions = "
+            f"      [i IN dp_keep | n.description_parts_parse_versions[i]] + $new_parse_versions, "
+            f"    n.gmt_updated = datetime()"
+        )
+        async with self._session() as session:
+            await session.run(
+                query,
+                collection_id=self._collection_id,
+                name=target_name,
+                entity_type=last_entity_type,
+                strip_keys=strip_keys,
+                new_member_jsons=new_member_jsons,
+                new_part_jsons=new_part_jsons,
+                new_doc_ids=new_doc_ids,
+                new_parse_versions=new_parse_versions,
+            )
+
     async def upsert_relation_with_lineage(
         self,
         *,