From 4ab1a257a26fc0a5e215786589413de3ebda6822 Mon Sep 17 00:00:00 2001 From: David de Boer Date: Sat, 18 Jul 2026 15:50:34 +0200 Subject: [PATCH 1/4] feat(search)!: project per root type over caller-given roots Retire the whole-schema projection: projectRoots is now the only projection entry point, projecting one root type over the roots the caller (the pipeline selector) supplies rather than discovering them from rdf:type. projectGraph, frameByType and rootsByType go, and buildSubjectIndex drops its rootTypes parameter and no longer scans for type triples. TypedSearchDocument moves to @lde/search-pipeline, where a terminal routes; @lde/search now yields a bare SearchDocument and stays pipeline-free. BREAKING CHANGE: @lde/search no longer exports projectGraph or TypedSearchDocument, and buildSubjectIndex takes only the quad source (no rootTypes). --- packages/search/README.md | 72 ++++++------ packages/search/package.json | 2 +- packages/search/src/engine.ts | 18 +-- packages/search/src/frame-by-type.ts | 73 ++++-------- packages/search/src/index.ts | 10 +- packages/search/src/project.ts | 62 ++--------- packages/search/test/frame-by-type.test.ts | 124 +++++++++------------ packages/search/test/project.test.ts | 69 +----------- packages/search/vite.config.ts | 2 +- 9 files changed, 135 insertions(+), 297 deletions(-) diff --git a/packages/search/README.md b/packages/search/README.md index 34f85a6f..43e691d9 100644 --- a/packages/search/README.md +++ b/packages/search/README.md @@ -29,11 +29,11 @@ It provides four things: out of it, so the two cannot drift; - **engine port** – `SearchEngine` and the logical result types (`SearchResult` / `SearchHit` / `ResultDocument` / `Reference` / …); -- **streaming projection** – `projectGraph`, RDF `CONSTRUCT` quads → flat - search documents. +- **streaming projection** – `projectRoots`, RDF `CONSTRUCT` quads → flat + search documents, one root type at a time. ``` -SearchSchema ─┬─► projection (projectGraph → flat documents) [here] +SearchSchema ─┬─► projection (projectRoots → flat documents) [here] ├─► engine adapter (collection definition + query compiler) e.g. @lde/search-typesense ├─► query semantics (SearchQuery, filter/sort/facet) [here] └─► API surface (GraphQL / REST) e.g. @lde/search-api-graphql @@ -58,7 +58,7 @@ a typed, deterministic function – easy to test, and swappable per deployment. Exports are stratified by audience: - **`@lde/search`** – the authoring surface: `defineSearchType`, - `searchSchema`, `projectGraph` (+ the IR readers `derive` functions use), + `searchSchema`, `projectRoots` (+ the IR readers `derive` functions use), validation, and every model/query/result type. - **`@lde/search/adapter`** – plumbing for engine adapters and API surfaces: `physicalFields`, the field selectors, `assertValidQuery`, the filter @@ -78,8 +78,9 @@ and GraphQL (one of the surfaces): | `SearchType` | One root type’s complete declaration: its logical API `name`, its `class` IRI and its fields (incl. derived) | NodeShape | object type | | `SearchSchema` | The whole search declaration: every `SearchType`, keyed by `class` IRI – build one with `searchSchema(...types)` | shapes graph | schema | -`projectGraph` and the GraphQL surface consume a `SearchSchema` (projecting -every type in one pass); the engine port executes one `SearchType` at a time. +`projectRoots` and the engine port each execute one `SearchType` at a time – +projection over the roots the pipeline selector supplied for that type; the +GraphQL surface consumes the whole `SearchSchema`. ### API conventions @@ -110,7 +111,7 @@ query compiler all share. ```ts import { defineSearchType, - projectGraph, + projectRoots, irisOf, searchSchema, } from '@lde/search'; @@ -150,14 +151,12 @@ const DATASET = defineSearchType({ ], }); -for await (const { searchType, document } of projectGraph( - quads, - searchSchema(DATASET), -)) { - // one flat search document per matching subject, streamed and tagged with the - // SearchType it was framed from — so a multi-collection writer can route each - // document to the collection for its type. A single-collection consumer just - // reads `document`. +const schema = searchSchema(DATASET); +for await (const document of projectRoots(quads, roots, schema, DATASET)) { + // one flat search document per given root subject, streamed. The caller (the + // pipeline selector) supplies `roots`; tagging a document with its type for a + // multi-collection writer is the pipeline glue’s job (see + // `@lde/search-pipeline`), not the projection’s. } ``` @@ -219,31 +218,38 @@ source fails at startup. A reference without a `labelSource` stays id-only. ## Projection -`projectGraph` is fully streaming: subjects are grouped and framed one at a time -and documents are yielded as produced, so beyond a subject index memory stays -flat at scale (framing the whole graph at once is roughly O(N²)). Duplicate -triples are collapsed first, because some SPARQL engines (e.g. QLever) do not -deduplicate `CONSTRUCT` output. The IR carries no `@context`, so a `derive` function -reading it sees full predicate IRIs with language tags preserved. - -Each yielded value is a `{ searchType, document }` pair: the whole schema -projects into **one** mixed stream, and every document is tagged with the -`SearchType` it was framed from. That tag is what lets the write side fan the -stream out to per-type collections without re-deriving the type from the -document – see `@lde/search-pipeline`’s multi-collection writer. - -`projectGraph` **consumes the quads once** – a single scan builds the subject -index every type frames off – and so accepts any `Iterable`, not just a -materialized array. A caller merging several sources can pass a chained +`projectRoots` projects **one root type** over the roots the caller supplies – +the pipeline selector already holds them, so nothing is discovered from +`rdf:type` and a `CONSTRUCT` need emit no type triple. It is fully streaming: +each root’s subgraph is framed one at a time and its document yielded as +produced, so beyond a subject index memory stays flat at scale (framing the +whole graph at once is roughly O(N²)). Duplicate triples are collapsed first, +because some SPARQL engines (e.g. QLever) do not deduplicate `CONSTRUCT` output. +The IR carries no `@context`, so a `derive` function reading it sees full +predicate IRIs with language tags preserved. `assertTypeInSchema` guards that +the passed `SearchType` is a member of the schema – the port’s own membership +check – so no schema is ever forged to scope a projection to one type. + +`projectRoots` yields a **bare** `SearchDocument`. Tagging each document with the +`SearchType` it belongs to, so the write side can fan a mixed stream out to +per-type collections, is a routing concern owned by the pipeline glue – see +`@lde/search-pipeline`’s `searchStages` and multi-collection writer – not the +projection. One stage per root type keeps `@lde/search` pipeline-free. + +`projectRoots` **consumes the quads once** – a single scan builds the subject +index the roots frame off – and so accepts any `Iterable`, not just a +materialized array. A caller merging several readers can pass a chained generator instead of building a third full array at the projection peak: ```ts -projectGraph( +projectRoots( (function* () { yield* registerQuads; yield* dkgQuads; })(), + roots, schema, + DATASET, ); ``` @@ -307,7 +313,7 @@ unrepresentable; the port enforces them for everyone else (deployment ## Typed results An engine is **bound to the whole `SearchSchema` at construction** – like -every other schema consumer (`projectGraph(quads, schema)`, +every other schema consumer (`projectRoots(quads, roots, schema, type)`, `buildGraphQLSchema(schema)`): the adapter factory takes the deployment’s declaration, so a query can never meet the wrong index, and deployment-level concerns (the label cache, cross-type search, facet batching) have one home. diff --git a/packages/search/package.json b/packages/search/package.json index 28319a13..0384f91d 100644 --- a/packages/search/package.json +++ b/packages/search/package.json @@ -39,12 +39,12 @@ "dependencies": { "@lde/text-normalization": "^0.1.1", "@rdfjs/types": "^2.0.1", - "@tpluscode/rdf-ns-builders": "^5.0.0", "inflection": "^3.0.2", "jsonld": "^9.0.0", "tslib": "^2.3.0" }, "devDependencies": { + "@tpluscode/rdf-ns-builders": "^5.0.0", "@types/jsonld": "^1.5.15", "n3": "^2.1.1" }, diff --git a/packages/search/src/engine.ts b/packages/search/src/engine.ts index 2326d1c2..279ab7bf 100644 --- a/packages/search/src/engine.ts +++ b/packages/search/src/engine.ts @@ -4,10 +4,10 @@ import type { SearchSchema, SearchType } from './schema.js'; /** * The engine port: the boundary a concrete engine adapter (e.g. the engine * `@lde/search-typesense`’s `createTypesenseSearchEngine` returns) implements. - * An engine is **bound to the whole {@link SearchSchema} at construction** — + * An engine is **bound to the whole {@link SearchSchema} at construction** – * the adapter factory takes the deployment’s declaration together with the * physical location of every type (its collections map), mirroring the other - * schema consumers (`projectGraph(quads, schema)`, + * schema consumers (`projectRoots(quads, roots, schema, type)`, * `buildGraphQLSchema(schema)`): everything is a function of the schema. A * query can never meet the wrong index, deployment-level concerns (label * cache, cross-type search, facet batching) have one home, and a search names @@ -20,8 +20,8 @@ import type { SearchSchema, SearchType } from './schema.js'; * * Port contract: an adapter ALWAYS rejects a `searchType` that is not in its * bound schema, and ALWAYS validates the incoming query against it - * (`assertValidQuery`) — unknown or non-filterable fields, mismatched - * operators, unknown facets — rather than passing garbage to its engine. + * (`assertValidQuery`) – unknown or non-filterable fields, mismatched + * operators, unknown facets – rather than passing garbage to its engine. * Validation is not the caller’s job: it must hold for every surface and for * injected deployment policy. * @@ -34,7 +34,7 @@ import type { SearchSchema, SearchType } from './schema.js'; export interface SearchEngine< Types extends readonly SearchType[] = readonly SearchType[], > { - /** The declaration this engine serves — exposed so a surface can route and + /** The declaration this engine serves – exposed so a surface can route and * a caller can enumerate the searchable types. */ readonly schema: SearchSchema; search( @@ -94,7 +94,7 @@ export type FacetMap = Readonly< >; /** - * The facet-field-name union of a search type — the keys a {@link SearchResult}’s + * The facet-field-name union of a search type – the keys a {@link SearchResult}’s * `facets` can hold. Narrow for a declaration captured as a literal (via * `defineSearchType` or `as const satisfies SearchType`); a plain * `: SearchType` annotation degrades to `string` (its field names are not @@ -105,7 +105,7 @@ export type FacetFieldsOf = SearchType extends Type : Extract['name']; /** - * The output-field-name union of a search type — the keys a {@link ResultDocument} + * The output-field-name union of a search type – the keys a {@link ResultDocument} * can hold. Like {@link FacetFieldsOf}, degrades to `string` for a widened * declaration. */ @@ -116,7 +116,7 @@ export type OutputFieldsOf = SearchType extends Type /** * One result row. `id` (the stable document key, an IRI) is kept *out* of * {@link ResultDocument}: it is always present and is the hit’s identity, a - * different contract from the optional, typed logical field values — and it maps + * different contract from the optional, typed logical field values – and it maps * straight onto the GraphQL output’s guaranteed `id: String!`. The document * holds only the selectable fields. */ @@ -126,7 +126,7 @@ export interface SearchHit { } /** - * The logical result document at the query seam — engine- and RDF-neutral. + * The logical result document at the query seam – engine- and RDF-neutral. * Distinct from the flat, fanned-out projection `SearchDocument` that lives * index-side: this carries logical fields with language maps and references, * ready for a surface to shape. Keyed by output field name; `Partial` because a diff --git a/packages/search/src/frame-by-type.ts b/packages/search/src/frame-by-type.ts index b4f6d1ca..abca6ee6 100644 --- a/packages/search/src/frame-by-type.ts +++ b/packages/search/src/frame-by-type.ts @@ -1,8 +1,5 @@ import type { Quad } from '@rdfjs/types'; import jsonld from 'jsonld'; -import { rdf } from '@tpluscode/rdf-ns-builders'; - -const RDF_TYPE = rdf.type.value; /** A framed JSON-LD node (full-IRI keys); the engine-agnostic search IR. */ export type FramedNode = Record; @@ -10,38 +7,31 @@ export type FramedNode = Record; const FRAME_OPTIONS = { omitGraph: false, embed: '@always' as const }; /** - * A one-pass index of a quad source: every subject’s (deduplicated) triples, - * plus the root subjects of each requested type in appearance order. Built by - * {@link buildSubjectIndex} from a single iteration of the source, so a caller - * can pass a chained generator (`function* () { yield* a; yield* b; }`) rather - * than materialize a merged array; a multi-type projection then frames every - * type off this one index instead of re-scanning per type. + * A one-pass index of a quad source: every subject’s (deduplicated) triples. + * Built by {@link buildSubjectIndex} from a single iteration of the source, so a + * caller can pass a chained generator (`function* () { yield* a; yield* b; }`) + * rather than materialize a merged array, then frame any of its subjects off + * this one index. */ export interface SubjectIndex { /** Each subject IRI → its own deduplicated triples, in first-seen order. */ readonly bySubject: ReadonlyMap; - /** Each requested root type IRI → its root subjects, in appearance order. */ - readonly rootsByType: ReadonlyMap; } /** - * Iterate `quads` once, grouping each subject’s triples and collecting the root - * subjects of every type in `rootTypes`. Duplicate triples are collapsed here - * because some SPARQL engines (e.g. QLever) do not deduplicate CONSTRUCT - * output. The source is consumed a single time, so it may be a one-shot - * iterable (a generator chaining several sources); the whole subject index is - * held, but never more than that plus one framed subgraph at a time + * Iterate `quads` once, grouping each subject’s triples. Duplicate triples are + * collapsed here because some SPARQL engines (e.g. QLever) do not deduplicate + * CONSTRUCT output. The source is consumed a single time, so it may be a + * one-shot iterable (a generator chaining several sources); the whole subject + * index is held, but never more than that plus one framed subgraph at a time * downstream. + * + * Roots are never discovered here – the caller ({@link projectRoots}, from the + * pipeline selector) already holds them and passes them to {@link frameSubjects} + * – so `quads` need carry no `rdf:type` triple. */ -export function buildSubjectIndex( - quads: Iterable, - rootTypes: Iterable, -): SubjectIndex { +export function buildSubjectIndex(quads: Iterable): SubjectIndex { const bySubject = new Map(); - const rootsByType = new Map(); - for (const type of rootTypes) { - rootsByType.set(type, []); - } const seen = new Set(); for (const quad of quads) { const key = `${quad.subject.value} ${quad.predicate.value} ${quad.object.value} ${quad.object.termType === 'Literal' ? quad.object.language || quad.object.datatype.value : ''}`; @@ -56,16 +46,8 @@ export function buildSubjectIndex( } else { owned.push(quad); } - // `rootsByType` is keyed by exactly the requested root types, so the lookup - // doubles as the membership test: a hit means this subject is a root. - if (quad.predicate.value === RDF_TYPE) { - const roots = rootsByType.get(quad.object.value); - if (roots !== undefined) { - roots.push(subject); - } - } } - return { bySubject, rootsByType }; + return { bySubject }; } /** @@ -76,11 +58,9 @@ export function buildSubjectIndex( * is held (whole-graph `jsonld.frame()` is ~O(N²)). The frame carries no * `@context`, so framed keys are full predicate IRIs. * - * The roots are supplied explicitly rather than discovered from `rdf:type`: a - * caller with the roots already in hand ({@link projectRoots}, from the pipeline - * selector) passes them directly; a whole-graph caller ({@link frameByType}, - * {@link projectGraph}) passes {@link SubjectIndex.rootsByType}. A root absent - * from the index simply frames nothing. + * The roots are supplied explicitly rather than discovered from `rdf:type`: the + * caller ({@link projectRoots}, from the pipeline selector) already holds them + * and passes them directly. A root absent from the index simply frames nothing. */ export async function* frameSubjects( index: SubjectIndex, @@ -114,18 +94,3 @@ export async function* frameSubjects( } } } - -/** - * Frame CONSTRUCT quads into one JSON-LD node per subject of `rootType`, for a - * single root type. Consumes `quads` once (see {@link buildSubjectIndex}), so - * it accepts any `Iterable` – a materialized array or a chained generator. For - * a multi-type schema, {@link buildSubjectIndex} + {@link frameSubjects} share - * one index across types rather than re-scanning per type. - */ -export async function* frameByType( - quads: Iterable, - rootType: string, -): AsyncIterable { - const index = buildSubjectIndex(quads, [rootType]); - yield* frameSubjects(index, index.rootsByType.get(rootType) ?? []); -} diff --git a/packages/search/src/index.ts b/packages/search/src/index.ts index 0132902b..09eb36ba 100644 --- a/packages/search/src/index.ts +++ b/packages/search/src/index.ts @@ -7,14 +7,8 @@ // Projection: RDF CONSTRUCT quads → flat search documents, driven by the // unified SearchField/SearchType model. The IR readers (irisOf, …) are here // because `derive` functions are written against them. -export { - projectGraph, - projectRoots, - irisOf, - literalsOf, - firstLiteralOf, -} from './project.js'; -export type { SearchDocument, TypedSearchDocument } from './project.js'; +export { projectRoots, irisOf, literalsOf, firstLiteralOf } from './project.js'; +export type { SearchDocument } from './project.js'; // Unified field model: one declaration drives projection, engine collection // schema, query semantics and the GraphQL surface – a discriminated union by diff --git a/packages/search/src/project.ts b/packages/search/src/project.ts index a9340d95..62aba77b 100644 --- a/packages/search/src/project.ts +++ b/packages/search/src/project.ts @@ -21,17 +21,6 @@ import { /** A flat search document. `id` is the engine document key. */ export type SearchDocument = { id: string } & Record; -/** - * A projected document tagged with the {@link SearchType} it was projected - * from – one element of {@link projectGraph}’s mixed, whole-schema stream. The - * tag is what lets a multi-collection writer route each document to the - * collection for its type without re-deriving the type from the document. - */ -export interface TypedSearchDocument { - readonly searchType: SearchType; - readonly document: SearchDocument; -} - /** * Project one framed JSON-LD node into a flat search document: apply each field * of the type in declaration order. A field with a `derive` function computes @@ -57,47 +46,15 @@ export function projectDocument( return document; } -/** - * Frame `quads` for every root type in the schema and project each node with its - * type’s declaration – the multi-shape pipeline. Yields each document tagged - * with its {@link SearchType} ({@link TypedSearchDocument}), so a downstream - * multi-collection writer can route it to that type’s collection; a - * single-collection consumer just reads `.document`. Streams one document at a - * time so memory stays flat. The IR maps to a declaration by type, so adding a - * shape is adding a `SearchType` to the schema (no engine change). - * - * Consumes `quads` once (a single scan builds the shared subject index that - * every type frames off), so it accepts any `Iterable` – a materialized array or - * a chained generator merging several sources (`function* () { yield* a; yield* b; }`) - * with no intermediate copy at the projection peak. - */ -export async function* projectGraph( - quads: Iterable, - schema: SearchSchema, -): AsyncIterable { - const types = [...schema.values()]; - const index = buildSubjectIndex( - quads, - types.map((searchType) => searchType.class), - ); - for (const searchType of types) { - const roots = index.rootsByType.get(searchType.class) ?? []; - for await (const node of frameSubjects(index, roots)) { - yield { searchType, document: projectDocument(node, searchType) }; - } - } -} - /** * Project a single type over a known set of `roots` – the per-type, roots-given - * projection. Unlike {@link projectGraph}, the roots are supplied by the caller - * (the pipeline selector) rather than discovered from `rdf:type`, so `quads` - * need carry no type triples and the projection frames each distinct subject - * once. {@link assertTypeInSchema} guards that `searchType` belongs to `schema` - * – the port’s own membership check – so no schema is ever forged to scope a - * projection to one type. Yields a bare {@link SearchDocument}: tagging a - * document with its type is a routing concern, owned by the pipeline glue, not - * the projection. + * projection. The roots are supplied by the caller (the pipeline selector) + * rather than discovered from `rdf:type`, so `quads` need carry no type triples + * and the projection frames each distinct subject once. {@link assertTypeInSchema} + * guards that `searchType` belongs to `schema` – the port’s own membership check + * – so no schema is ever forged to scope a projection to one type. Yields a bare + * {@link SearchDocument}: tagging a document with its type is a routing concern, + * owned by the pipeline glue, not the projection. * * Consumes `quads` once, so it accepts any `Iterable` – a batch’s materialized * array or a chained generator merging several readers. @@ -109,12 +66,11 @@ export async function* projectRoots( searchType: SearchType, ): AsyncIterable { assertTypeInSchema(schema, searchType); - const index = buildSubjectIndex(quads, []); + const index = buildSubjectIndex(quads); // Distinct roots only. A selector may return an IRI more than once – a // non-`DISTINCT` `SELECT` over a one-to-many join yields the same subject per // matched row – and a repeated root would otherwise frame and emit a - // duplicate document under the same `id`. (`projectGraph`’s roots come from - // `rootsByType`, which is inherently unique, so it never needs this.) + // duplicate document under the same `id`. for await (const node of frameSubjects(index, [...new Set(roots)])) { yield projectDocument(node, searchType); } diff --git a/packages/search/test/frame-by-type.test.ts b/packages/search/test/frame-by-type.test.ts index 2fb3b26d..f38d8eca 100644 --- a/packages/search/test/frame-by-type.test.ts +++ b/packages/search/test/frame-by-type.test.ts @@ -3,7 +3,6 @@ import { Parser } from 'n3'; import { dcat, dcterms, foaf, rdf } from '@tpluscode/rdf-ns-builders'; import { buildSubjectIndex, - frameByType, frameSubjects, type FramedNode, } from '../src/frame-by-type.js'; @@ -24,19 +23,24 @@ async function collect( return out; } -describe('frameByType', () => { - it('frames each root subject’s one-hop subgraph into an IR node', async () => { +/** Build a subject index over `ntriples` and frame the given `roots`. */ +function frame(ntriples: string, roots: string[]): AsyncIterable { + return frameSubjects(buildSubjectIndex(quads(ntriples)), roots); +} + +describe('frameSubjects', () => { + it('frames each given root’s one-hop subgraph into an IR node', async () => { + // The roots are supplied by the caller (the pipeline selector); no rdf:type + // triple is needed to discover them. const nodes = await collect( - frameByType( - quads(` - <${rdf.type.value}> <${DATASET}> . + frame( + ` <${dcterms.title.value}> "Titel"@nl . <${dcterms.publisher.value}> . <${foaf.name.value}> "Org"@nl . - <${rdf.type.value}> <${DATASET}> . <${dcterms.title.value}> "Andere"@nl . - `), - DATASET, + `, + ['https://ex/d/1', 'https://ex/d/2'], ), ); @@ -55,37 +59,34 @@ describe('frameByType', () => { it('embeds one-hop blank-node references', async () => { const nodes = await collect( - frameByType( - quads(` - <${rdf.type.value}> <${DATASET}> . + frame( + ` <${dcat.distribution.value}> _:dist . _:dist <${dcterms.title.value}> "Distributie"@nl . - `), - DATASET, + `, + ['https://ex/d/1'], ), ); expect(nodes).toHaveLength(1); - expect(nodes[0]![dcat.distribution.value]).toMatchObject({ + expect(nodes[0][dcat.distribution.value]).toMatchObject({ [dcterms.title.value]: { '@language': 'nl', '@value': 'Distributie' }, }); }); - it('keeps a root subject that references another root-typed subject', async () => { - // A one-hop reference can itself be of the root type (e.g. a terminology + it('keeps a root that references another framed root', async () => { + // A one-hop reference can itself be a requested root (e.g. a terminology // source that is also a separately registered dataset). The referencing - // subject must still be projected – and the referenced one must not be - // emitted twice – even though both match the frame’s `@type`. + // subject must still be framed – and the referenced one must not be emitted + // twice – even though `frame()` embeds the referent inline as well. const nodes = await collect( - frameByType( - quads(` - <${rdf.type.value}> <${DATASET}> . + frame( + ` <${dcterms.title.value}> "Verwijzer"@nl . <${dcterms.source.value}> . - <${rdf.type.value}> <${DATASET}> . <${dcterms.title.value}> "Bron"@nl . - `), - DATASET, + `, + ['https://ex/d/1', 'https://ex/d/2'], ), ); @@ -101,82 +102,65 @@ describe('frameByType', () => { it('dedupes triples a CONSTRUCT may emit more than once', async () => { const nodes = await collect( - frameByType( - quads(` + frame( + ` <${rdf.type.value}> <${DATASET}> . <${rdf.type.value}> <${DATASET}> . <${dcterms.title.value}> "Titel"@nl . <${dcterms.title.value}> "Titel"@nl . - `), - DATASET, + `, + ['https://ex/d/1'], ), ); expect(nodes).toHaveLength(1); - expect(nodes[0]![dcterms.title.value]).toEqual({ + expect(nodes[0][dcterms.title.value]).toEqual({ '@language': 'nl', '@value': 'Titel', }); }); - it('yields nothing when there are no subjects of the root type', async () => { - const nodes = await collect( - frameByType( - quads(` <${foaf.name.value}> "Org"@nl .`), - DATASET, - ), - ); - expect(nodes).toEqual([]); - }); -}); - -describe('buildSubjectIndex / frameSubjects', () => { - it('scans a one-shot iterable once, indexing every requested root type', async () => { - const other = 'http://example.org/Other'; + it('scans a one-shot iterable once, framing every given root off one index', async () => { const source = quads(` - <${rdf.type.value}> <${DATASET}> . <${dcterms.title.value}> "Titel"@nl . - <${rdf.type.value}> <${other}> . <${dcterms.title.value}> "Ander"@nl . `); function* once(): Generator<(typeof source)[number]> { yield* source; } - // A single pass over the generator must still index both types. - const index = buildSubjectIndex(once(), [DATASET, other]); - const datasets = await collect( - frameSubjects(index, index.rootsByType.get(DATASET) ?? []), - ); - const others = await collect( - frameSubjects(index, index.rootsByType.get(other) ?? []), - ); + // A generator is exhausted after one pass; framing both roots off the single + // index proves the one-shot source is scanned exactly once. + const index = buildSubjectIndex(once()); + const first = await collect(frameSubjects(index, ['https://ex/d/1'])); + const second = await collect(frameSubjects(index, ['https://ex/x/1'])); - expect(datasets.map((node) => node['@id'])).toEqual(['https://ex/d/1']); - expect(others.map((node) => node['@id'])).toEqual(['https://ex/x/1']); + expect(first.map((node) => node['@id'])).toEqual(['https://ex/d/1']); + expect(second.map((node) => node['@id'])).toEqual(['https://ex/x/1']); }); it('frames each explicitly given root, ignoring rdf:type', async () => { // No type triples at all: the roots are supplied directly, the way a // pipeline selector supplies them. - const index = buildSubjectIndex( - quads(` - <${dcterms.title.value}> "Een"@nl . - <${dcterms.title.value}> "Twee"@nl . - `), - [], + const framed = await collect( + frame( + ` + <${dcterms.title.value}> "Een"@nl . + <${dcterms.title.value}> "Twee"@nl . + `, + ['https://ex/d/1'], + ), ); - const framed = await collect(frameSubjects(index, ['https://ex/d/1'])); expect(framed.map((node) => node['@id'])).toEqual(['https://ex/d/1']); }); it('frames nothing for a root absent from the index', async () => { - const index = buildSubjectIndex( - quads(` <${rdf.type.value}> <${DATASET}> .`), - [DATASET], - ); - expect(await collect(frameSubjects(index, ['urn:unregistered']))).toEqual( - [], - ); + expect( + await collect( + frame(` <${rdf.type.value}> <${DATASET}> .`, [ + 'urn:unregistered', + ]), + ), + ).toEqual([]); }); }); diff --git a/packages/search/test/project.test.ts b/packages/search/test/project.test.ts index c8b12bc1..96f9a498 100644 --- a/packages/search/test/project.test.ts +++ b/packages/search/test/project.test.ts @@ -1,9 +1,8 @@ import { describe, expect, it } from 'vitest'; import { Parser } from 'n3'; -import { dcat, dcterms, rdf, xsd } from '@tpluscode/rdf-ns-builders'; +import { dcat, dcterms, xsd } from '@tpluscode/rdf-ns-builders'; import { projectDocument, - projectGraph, projectRoots, irisOf, type SearchDocument, @@ -529,72 +528,6 @@ describe('projectDocument', () => { }); }); -describe('projectGraph', () => { - it('frames each root type in the schema and projects matching nodes', async () => { - const quads = new Parser({ format: 'N-Triples' }).parse(` - <${rdf.type.value}> <${DATASET}> . - <${dcterms.title.value}> "Titel"@nl . - <${rdf.type.value}> <${DATASET}> . - <${dcterms.title.value}> "Andere"@nl . - <${rdf.type.value}> . - <${dcterms.title.value}> "Ignored"@nl . - `); - - const documents: SearchDocument[] = []; - for await (const { searchType, document } of projectGraph( - quads, - searchSchema({ name: 'Dataset', class: DATASET, fields }), - )) { - expect(searchType.name).toBe('Dataset'); - documents.push(document); - } - - const ids = documents.map((document) => document.id).sort(); - expect(ids).toEqual(['https://ex/d/1', 'https://ex/d/2']); - const byId = Object.fromEntries( - documents.map((document) => [document.id, document]), - ); - expect(byId['https://ex/d/1'].title_search_nl).toBe('titel'); - expect(byId['https://ex/d/2'].title_nl).toBe('Andere'); - }); - - it('consumes a one-shot iterable once, framing every type off a single scan', async () => { - const other = 'http://example.org/Other'; - const quads = new Parser({ format: 'N-Triples' }).parse(` - <${rdf.type.value}> <${DATASET}> . - <${dcterms.title.value}> "Titel"@nl . - <${rdf.type.value}> <${other}> . - <${dcterms.title.value}> "Ander"@nl . - `); - // A generator is exhausted after one pass, so a per-type re-scan would drop - // every type but the first – projecting both proves the single-scan index. - function* once(): Generator<(typeof quads)[number]> { - yield* quads; - } - - const tagged: { type: string; id: string }[] = []; - for await (const { searchType, document } of projectGraph( - once(), - searchSchema( - { name: 'Dataset', class: DATASET, fields }, - { name: 'Other', class: other, fields }, - ), - )) { - tagged.push({ type: searchType.class, id: document.id }); - } - - expect(tagged.map((entry) => entry.id).sort()).toEqual([ - 'https://ex/d/1', - 'https://ex/x/1', - ]); - // Each document carries the type it was framed from, so a writer can route - // it to that type’s collection. - expect( - Object.fromEntries(tagged.map((entry) => [entry.id, entry.type])), - ).toEqual({ 'https://ex/d/1': DATASET, 'https://ex/x/1': other }); - }); -}); - describe('projectRoots', () => { const dataset = defineSearchType({ name: 'Dataset', class: DATASET, fields }); const schema = searchSchema(dataset); diff --git a/packages/search/vite.config.ts b/packages/search/vite.config.ts index c1151477..e59cd028 100644 --- a/packages/search/vite.config.ts +++ b/packages/search/vite.config.ts @@ -12,7 +12,7 @@ export default mergeConfig( thresholds: { functions: 100, lines: 100, - branches: 97.32, + branches: 98.14, statements: 100, }, }, From fd7e6348bcf6e6c71b665ce2c0fb4b1405507082 Mon Sep 17 00:00:00 2001 From: David de Boer Date: Sat, 18 Jul 2026 15:52:22 +0200 Subject: [PATCH 2/4] feat(search-pipeline)!: per-type projecting stages, retire the buffering writer Compose a search pipeline as one terminal and N per-type stages. searchStages builds one projecting Stage per root type: each selects its roots, extracts each root's quads, and projects the root-complete batch (projectRoots) into documents tagged with their SearchType, so memory is bounded by batchSize roots, not the dataset. selectByClass is a convenience selector for the object grain. searchIndexWriter becomes a Writer: it keeps ADR 9's per-collection fan-out and run lifecycle but stops projecting and stops buffering, routing each tagged document straight to its type's engine run. TypedSearchDocument now lives here, the glue that needs it. BREAKING CHANGE: searchIndexWriter now consumes TypedSearchDocument, not Quad, and no longer projects; compose it with searchStages. --- package-lock.json | 4 +- packages/search-api-graphql/README.md | 18 +- packages/search-pipeline/README.md | 98 +++++-- packages/search-pipeline/package.json | 2 +- packages/search-pipeline/src/index.ts | 3 + .../src/search-index-writer.ts | 182 ++++++------ packages/search-pipeline/src/search-stages.ts | 137 +++++++++ .../src/typed-search-document.ts | 23 ++ .../test/multi-collection.integration.test.ts | 56 ++-- .../test/search-index-writer.test.ts | 156 ++++++----- .../test/search-stages.test.ts | 261 ++++++++++++++++++ 11 files changed, 720 insertions(+), 220 deletions(-) create mode 100644 packages/search-pipeline/src/search-stages.ts create mode 100644 packages/search-pipeline/src/typed-search-document.ts create mode 100644 packages/search-pipeline/test/search-stages.test.ts diff --git a/package-lock.json b/package-lock.json index 6c21964a..e46c0d38 100644 --- a/package-lock.json +++ b/package-lock.json @@ -37872,12 +37872,12 @@ "dependencies": { "@lde/text-normalization": "^0.1.1", "@rdfjs/types": "^2.0.1", - "@tpluscode/rdf-ns-builders": "^5.0.0", "inflection": "^3.0.2", "jsonld": "^9.0.0", "tslib": "^2.3.0" }, "devDependencies": { + "@tpluscode/rdf-ns-builders": "^5.0.0", "@types/jsonld": "^1.5.15", "n3": "^2.1.1" }, @@ -37907,11 +37907,11 @@ "license": "MIT", "dependencies": { "@lde/search": "^0.8.0", - "@rdfjs/types": "^2.0.1", "tslib": "^2.3.0" }, "devDependencies": { "@lde/search-typesense": "^0.10.0", + "@rdfjs/types": "^2.0.1", "n3": "^2.1.1", "testcontainers": "^12.0.3", "typesense": "^3.0.6" diff --git a/packages/search-api-graphql/README.md b/packages/search-api-graphql/README.md index 8765bf36..069c2bad 100644 --- a/packages/search-api-graphql/README.md +++ b/packages/search-api-graphql/README.md @@ -3,21 +3,21 @@ The GraphQL surface for the [`@lde/search`](../search) core. **Both engine- and domain-agnostic:** it builds an executable [graphql-js](https://graphql.org/graphql-js/) `GraphQLSchema` from your whole -[`SearchSchema`](../search/README.md#terminology) at runtime — one root query +[`SearchSchema`](../search/README.md#terminology) at runtime – one root query field per `SearchType`, each searchable in its own way. All root fields are served by the same resolver implementation (no per-type code, no codegen); each root field gets its own instance of it, bound to that field’s `SearchType`, over any `SearchEngine`. It names neither your **domain** (each type’s GraphQL name -is the `SearchType`’s own logical `name` — `Dataset`, `Person`, `CreativeWork`, +is the `SearchType`’s own logical `name` – `Dataset`, `Person`, `CreativeWork`, …) nor your **engine** (the resolver calls the schema-bound `context.engine`, be it [`@lde/search-typesense`](../search-typesense) or another adapter). ## Runtime configuration, not codegen `buildGraphQLSchema(schema)` constructs the GraphQL schema once at startup from -the field model — no SDL artifact, no generated resolver stubs. For you that +the field model – no SDL artifact, no generated resolver stubs. For you that means: no codegen step in the build, no generated files to commit and review, -and no stale artifact that can drift from the declaration — change the +and no stale artifact that can drift from the declaration – change the `SearchType`, restart, and the API is current. (The flip side, no artifact showing contract changes as diffs, is restored by the [snapshot guard](#guarding-the-contract).) The field model @@ -67,8 +67,8 @@ only part of what you index, narrow the **schema argument** (`searchSchema(…)` is a cheap constructor): ```ts -// Index all three types… -projectGraph(quads, searchSchema(DATASET, PERSON, INTERNAL)); +// Index all three types (a projecting stage per type, see @lde/search-pipeline)… +searchSchema(DATASET, PERSON, INTERNAL); // …but serve only two. const gqlSchema = buildGraphQLSchema(searchSchema(DATASET, PERSON)); @@ -98,7 +98,7 @@ const gqlSchema = buildGraphQLSchema(searchSchema(DATASET, PERSON)); ## Guarding the contract Why the API, the index and a future REST surface cannot drift apart is the -search family’s overall approach — one field model, one query IR — described +search family’s overall approach – one field model, one query IR – described in [`@lde/search`](../search/README.md). Specific to this surface: the GraphQL contract is **frozen** (breaking to change), yet generated rather than handwritten, so nothing in the repo shows a contract change as a reviewable @@ -114,8 +114,8 @@ it('keeps the public GraphQL contract stable', () => { ``` The first run writes the emitted SDL to a committed snapshot file; every later -run re-emits and diffs against it. Any contract change — your own schema edit, +run re-emits and diffs against it. Any contract change – your own schema edit, or a new version of this library emitting different GraphQL for the same -declaration — fails the test and shows the SDL diff, until you consciously +declaration – fails the test and shows the SDL diff, until you consciously accept it (`vitest -u`) and the reviewer sees the contract change spelled out in the PR. diff --git a/packages/search-pipeline/README.md b/packages/search-pipeline/README.md index 2a6c62cf..7d4941c4 100644 --- a/packages/search-pipeline/README.md +++ b/packages/search-pipeline/README.md @@ -6,7 +6,8 @@ and a search engine’s document world, so indexing reuses the existing spine Discovery → Selection → Iteration → change-gate (skip-unchanged) – instead of rebuilding it per consumer. -The division of labour ([ADR 6](../../docs/decisions/0006-make-the-writer-transaction-aware.md)): +The division of labour ([ADR 6](../../docs/decisions/0006-make-the-writer-transaction-aware.md), +[ADR 13](../../docs/decisions/0013-project-inside-the-batch-per-root-type.md)): - [`@lde/search`](../search) owns the engine-agnostic projection (framed document + field model) and stays pipeline-free; @@ -14,18 +15,32 @@ The division of labour ([ADR 6](../../docs/decisions/0006-make-the-writer-transa is the engine adapter: a transactional, **single-collection** `Writer` consuming projected documents, owning the run lifecycle (Blue/green alias swap or In-place sweeps, cross-pod lock); -- **this package** composes them: `searchIndexWriter` is the one type-changing - step (quad → document), shared across engines, and – since one schema declares - several root types – it fans each document out to the engine writer for **its** - type’s collection ([ADR 9](../../docs/decisions/0009-route-a-whole-schema-projection-to-per-type-collections.md)). +- **this package** composes them: + - `searchStages` builds one projecting `Stage` per root type. Each stage + selects its own roots, extracts each root’s quads, and projects the + **root-complete batch** into documents tagged with their `SearchType` + (`TypedSearchDocument`) – so projection happens inside the batch and memory + is bounded by `batchSize` roots, not the dataset + ([ADR 13](../../docs/decisions/0013-project-inside-the-batch-per-root-type.md)); + - `searchIndexWriter` is the pipeline’s single terminal: an engine-agnostic + router that dispatches each tagged document to the engine writer for **its** + type’s collection ([ADR 9](../../docs/decisions/0009-route-a-whole-schema-projection-to-per-type-collections.md)). + It owns no projection and buffers nothing. + +So a search pipeline is **one terminal, N stages**: `new Pipeline({ datasetSelector, stages: searchStages(...), writers: searchIndexWriter(...) })`. ## Usage ```typescript -import { Pipeline, Stage, SparqlConstructReader } from '@lde/pipeline'; +import { Pipeline, SparqlConstructReader } from '@lde/pipeline'; import { searchSchema } from '@lde/search'; import { BlueGreenRebuild } from '@lde/search-typesense'; -import { searchIndexWriter } from '@lde/search-pipeline'; +import { + searchStages, + selectByClass, + searchIndexWriter, + type TypedSearchDocument, +} from '@lde/search-pipeline'; // One schema, several root types: the `datasets` catalog plus the Organization // label collection its references resolve against. @@ -60,14 +75,33 @@ const schema = searchSchema( }, ); -const pipeline = new Pipeline({ +const dataset = schema.get('http://www.w3.org/ns/dcat#Dataset')!; +const organization = schema.get('http://xmlns.com/foaf/0.1/Organization')!; + +const pipeline = new Pipeline({ datasetSelector, - stages: [ - new Stage({ - name: 'extract', - readers: new SparqlConstructReader({ query: '…' }), - }), - ], + // One stage per root type: each selects its own roots (here by class – + // `selectByClass` is a convenience for the object grain; a deployment writes + // its own selector where the entry point is a domain fact) and extracts each + // root’s quads. `rootVariable` couples the selector’s projected variable to + // what the projection reads; it must not be `dataset` (reserved by the reader). + stages: searchStages({ + schema, + types: [ + { + searchType: dataset, + rootVariable: 'root', + itemSelector: selectByClass(dataset), + readers: new SparqlConstructReader({ query: '…' }), + }, + { + searchType: organization, + rootVariable: 'root', + itemSelector: selectByClass(organization), + readers: new SparqlConstructReader({ query: '…' }), + }, + ], + }), // Each type gets its own collection – an independent blue/green rebuild – // named by the adapter from the type itself (`Dataset` → `datasets`, // `Organization` → `organizations`), so no naming map is passed here and the @@ -101,14 +135,21 @@ Whatever the writers are named, the engine reading these collections must be given the same names (`collections`), or it reads the derived ones and finds an empty index. -Each dataset’s extracted CONSTRUCT quads are buffered until the dataset -completes, then projected once over the **whole** schema (`@lde/search`’s -`projectGraph`) into one mixed, type-tagged stream, and each document is -dispatched to the engine run for its type’s collection – before the engine acts -on the dataset’s completion, so an In-place stale sweep never races its own -documents. The run lifecycle (run context, per-dataset flush outcome, -commit/abort) passes through unchanged: the engine writer’s update mode governs, -and the pipeline never branches on it, nor on how many collections there are. +Each stage projects its own root-complete batches (`@lde/search`’s +`projectRoots`) into documents tagged with its `SearchType`, and the terminal +dispatches each to the engine run for its type’s collection as it arrives – +before the engine acts on the dataset’s completion, so an In-place stale sweep +never races its own documents. The run lifecycle (run context, per-dataset flush +outcome, commit/abort) passes through unchanged: the engine writer’s update mode +governs, and the pipeline never branches on it, nor on how many collections +there are. + +**Root selection is the deployment’s contract.** A selector that does not bind +the CONSTRUCT’s subject yields a batch that is not root-complete, and projection +will then silently emit partial documents – a promise `@lde/pipeline` cannot +check ([ADR 13](../../docs/decisions/0013-project-inside-the-batch-per-root-type.md)). +`selectByClass(searchType)` is a convenience for the object grain (where the +type’s `class` really is the source class), **not** a default. ## Per-collection isolation @@ -130,7 +171,12 @@ See [ADR 9](../../docs/decisions/0009-route-a-whole-schema-projection-to-per-typ ## Memory -Memory is bounded by one dataset’s extraction and released at each dataset -flush – nothing accumulates across datasets. Streaming bounded entity batches -_within_ one huge dataset needs the two-level iteration (dataset → -entity-URI batches) and is not implemented yet. +Memory is bounded by **`batchSize` roots**, not by the input +([ADR 12](../../docs/decisions/0012-bound-memory-by-the-unit-of-work-not-the-input.md), +[ADR 13](../../docs/decisions/0013-project-inside-the-batch-per-root-type.md)). +Projection runs inside each root-complete batch and streams straight through the +terminal to the engine run, which holds only its own batch – nothing accumulates +across a dataset or across datasets, so the same pipeline indexes a 15-quad +catalog entry and a dataset of millions of objects. The one irreducible atom is a +single root’s quads: unbounded for a pathological root, and nothing here changes +that. diff --git a/packages/search-pipeline/package.json b/packages/search-pipeline/package.json index 74a25e03..0578b98b 100644 --- a/packages/search-pipeline/package.json +++ b/packages/search-pipeline/package.json @@ -26,11 +26,11 @@ ], "dependencies": { "@lde/search": "^0.8.1", - "@rdfjs/types": "^2.0.1", "tslib": "^2.3.0" }, "devDependencies": { "@lde/search-typesense": "^0.10.2", + "@rdfjs/types": "^2.0.1", "n3": "^2.1.1", "testcontainers": "^12.0.3", "typesense": "^3.0.6" diff --git a/packages/search-pipeline/src/index.ts b/packages/search-pipeline/src/index.ts index 8530fdd4..0a611a18 100644 --- a/packages/search-pipeline/src/index.ts +++ b/packages/search-pipeline/src/index.ts @@ -1,2 +1,5 @@ +export { searchStages, selectByClass } from './search-stages.js'; +export type { SearchStagesOptions, SearchStageType } from './search-stages.js'; export { searchIndexWriter } from './search-index-writer.js'; export type { SearchIndexWriterOptions } from './search-index-writer.js'; +export type { TypedSearchDocument } from './typed-search-document.js'; diff --git a/packages/search-pipeline/src/search-index-writer.ts b/packages/search-pipeline/src/search-index-writer.ts index 3cb374cc..3fd295e8 100644 --- a/packages/search-pipeline/src/search-index-writer.ts +++ b/packages/search-pipeline/src/search-index-writer.ts @@ -1,24 +1,20 @@ -import type { Quad } from '@rdfjs/types'; import type { Dataset } from '@lde/dataset'; -import type { - DatasetOutcome, - RunContext, - RunWriter, - Writer, -} from '@lde/pipeline'; import { - projectGraph, - type SearchDocument, - type SearchSchema, - type SearchType, -} from '@lde/search'; + AsyncQueue, + type DatasetOutcome, + type RunContext, + type RunWriter, + type Writer, +} from '@lde/pipeline'; +import type { SearchDocument, SearchSchema, SearchType } from '@lde/search'; +import type { TypedSearchDocument } from './typed-search-document.js'; /** Options for {@link searchIndexWriter}. */ export interface SearchIndexWriterOptions { /** - * The declarative schema driving the projection: one {@link SearchType} per - * root type, each mapping framed RDF to flat search-document fields. The - * writer opens one engine run per type in it. + * The declarative schema: one {@link SearchType} per root type. The writer + * opens one engine run per type in it and routes each document to its type’s + * run. Must be the same schema the stages project through. */ schema: SearchSchema; /** @@ -31,20 +27,23 @@ export interface SearchIndexWriterOptions { * multi-collection deployment (the Dataset Register’s `datasets` plus its * Organization / Class / TerminologySource label collections) returns a * distinct writer per type, each an independent blue/green rebuild with its - * own collection, alias and cross-pod lock. The projection is whole-schema - * either way; the per-collection fan-out is this writer’s job. + * own collection, alias and cross-pod lock. The per-collection fan-out is this + * writer’s job. */ writerFor: (searchType: SearchType) => Writer; } /** - * The projection step of a search-indexing pipeline, as a quad `Writer`, fanned - * out across a type’s collections. It turns each dataset’s extracted CONSTRUCT - * quads into engine-agnostic search documents ({@link projectGraph}: frame by - * root type, then project each node with its type’s declaration) and dispatches - * each document to the engine run for **its** type. This is the one - * type-changing step (quad → document), shared across engines – which is why it - * lives here and not inside an engine adapter. + * The single terminal of a search-indexing pipeline: an engine-agnostic router + * that fans already-projected documents out across a type’s collections. The + * per-type {@link https://github.com/ldelements/lde/blob/main/docs/decisions/0013-project-inside-the-batch-per-root-type.md | stages} + * project inside the batch and tag each document with its {@link SearchType} + * ({@link TypedSearchDocument}); this writer dispatches each document to the + * engine run for **its** type by `searchType.class`. It **owns no projection** + * (that moved into the stages) and **buffers nothing** (documents stream through + * to the run as they arrive) – it is purely the per-collection fan-out + * {@link https://github.com/ldelements/lde/blob/main/docs/decisions/0009-route-a-whole-schema-projection-to-per-type-collections.md | ADR 9} + * made it. * * Each root type is an independent engine run (its own collection, alias and * lock), so the collections commit, sweep and fail in isolation: @@ -61,19 +60,16 @@ export interface SearchIndexWriterOptions { * - `abort` (a run failure, or a partial commit) drops every half-built * collection that has not committed and leaves the live ones untouched. * - * A dataset’s quads are buffered until its flush and projected then, so the - * documents land before every engine acts on the dataset’s completion (e.g. an - * In-place stale sweep). Memory is bounded by one dataset’s extraction, and - * released at each flush – nothing accumulates across datasets. Streaming - * bounded entity batches *within* one huge dataset needs the two-level - * iteration (dataset → entity-URI batches) and is not implemented yet. + * Memory is bounded by one batch of documents per type, not the dataset: + * `write` routes each document to its type’s run through a bounded queue and + * never accumulates them. */ export function searchIndexWriter( options: SearchIndexWriterOptions, -): Writer { +): Writer { const { schema, writerFor } = options; // One engine writer per root type, built once; each run opens them all. Keyed - // by the type IRI, which is also how a projected document names its type. + // by the type IRI, which is also how a tagged document names its type. const writers = new Map>( [...schema.values()].map((searchType) => [ searchType.class, @@ -82,7 +78,9 @@ export function searchIndexWriter( ); return { - async openRun(context: RunContext): Promise> { + async openRun( + context: RunContext, + ): Promise> { const runs = new Map>(); try { for (const [typeIri, writer] of writers) { @@ -101,62 +99,60 @@ export function searchIndexWriter( // (a committed blue/green rebuild’s abort drops its now-live collection). const committed = new Set(); - // One buffered pass per dataset, keyed by IRI. The pipeline processes - // datasets sequentially, but keeping the key explicit makes a write after - // another dataset's flush safe too. - const passes = new Map(); - - const project = async (pass: { - dataset: Dataset; - quads: Quad[]; - }): Promise => { - passes.delete(pass.dataset.iri.toString()); - if (pass.quads.length === 0) { - return; - } - // Whole-schema projection into one mixed stream, split by type here so - // each type’s documents reach its own collection’s run. - const byType = new Map(); - for await (const { searchType, document } of projectGraph( - pass.quads, - schema, - )) { - const documents = byType.get(searchType.class); - if (documents === undefined) { - byType.set(searchType.class, [document]); - } else { - documents.push(document); - } - } - // Every yielded type is a schema type, so it has a run; iterate the - // runs and write each the documents projected for its type (none, for a - // type absent from this pass). - for (const [typeIri, run] of runs) { - const documents = byType.get(typeIri); - if (documents !== undefined) { - await run.write(pass.dataset, stream(documents)); - } - } - }; - return { - write: async (dataset: Dataset, quads: AsyncIterable) => { - const key = dataset.iri.toString(); - let pass = passes.get(key); - if (pass === undefined) { - pass = { dataset, quads: [] }; - passes.set(key, pass); - } - for await (const quad of quads) { - pass.quads.push(quad); + write: async ( + dataset: Dataset, + items: AsyncIterable, + ) => { + // Route each tagged document to its type’s run, streaming. A stage + // writes one type, but the terminal carries no stage identity, so it + // routes per item by the tag. Each type gets one `run.write`, fed a + // bounded queue the run drains concurrently – so memory stays O(batch) + // per type, never O(dataset). Almost always one lane per call. + const lanes = new Map< + string, + { queue: AsyncQueue; done: Promise } + >(); + const laneFor = (searchType: SearchType) => { + let lane = lanes.get(searchType.class); + if (lane === undefined) { + const run = runs.get(searchType.class); + if (run === undefined) { + throw new Error( + `No engine run for search type “${searchType.name}” (${searchType.class}); it is not in this writer’s schema.`, + ); + } + const queue = new AsyncQueue(); + const done = run.write(dataset, queue); + // If the run stops consuming (its write rejects), unblock the + // producer so a full queue cannot deadlock the push loop below. + done.catch((error: unknown) => queue.abort(error)); + lane = { queue, done }; + lanes.set(searchType.class, lane); + } + return lane; + }; + + try { + for await (const { searchType, document } of items) { + await laneFor(searchType).queue.push(document); + } + for (const lane of lanes.values()) { + lane.queue.close(); + } + await Promise.all([...lanes.values()].map((lane) => lane.done)); + } catch (error) { + for (const lane of lanes.values()) { + lane.queue.abort(error); + } + await Promise.allSettled( + [...lanes.values()].map((lane) => lane.done), + ); + throw error; } }, flush: async (dataset: Dataset, outcome: DatasetOutcome) => { - const pass = passes.get(dataset.iri.toString()); - if (pass !== undefined) { - await project(pass); - } // Flush every collection independently: one collection’s flush failure // (a rollback, or an In-place stale sweep) must not skip another’s – // the pipeline isolates a flush error per dataset and still commits, @@ -170,22 +166,15 @@ export function searchIndexWriter( }, reset: async (dataset: Dataset) => { - // Discard the buffered pass so the re-run replaces it, and let every - // collection’s run discard whatever it already holds for the dataset – - // independently, so one collection’s reset failure never leaves - // another holding the discarded pass’s documents into the re-run. - passes.delete(dataset.iri.toString()); + // Let every collection’s run discard whatever it already holds for the + // dataset – independently, so one collection’s reset failure never + // leaves another holding the discarded documents into the re-run. await settleAll(runs.values(), 'reset', (run) => run.reset?.(dataset), ); }, commit: async () => { - // Safety net: project passes that were never flushed, so a committed - // run never silently drops written quads. - for (const pass of [...passes.values()]) { - await project(pass); - } // Commit every collection independently, so one failure neither // blocks nor wipes another – the `datasets` index goes live even if a // label collection cannot. Record each collection that went live, so @@ -237,8 +226,3 @@ async function settleAll( ); } } - -/** Yield the given documents as an async iterable, like a streaming source. */ -async function* stream(items: readonly Item[]): AsyncIterable { - yield* items; -} diff --git a/packages/search-pipeline/src/search-stages.ts b/packages/search-pipeline/src/search-stages.ts new file mode 100644 index 00000000..82cb905b --- /dev/null +++ b/packages/search-pipeline/src/search-stages.ts @@ -0,0 +1,137 @@ +import { + SparqlItemSelector, + Stage, + type ItemSelector, + type StageReaders, +} from '@lde/pipeline'; +import { projectRoots, type SearchSchema, type SearchType } from '@lde/search'; +import type { TypedSearchDocument } from './typed-search-document.js'; + +/** One root type’s stage in a search pipeline. */ +export interface SearchStageType { + /** + * The {@link SearchType} this stage projects. Must belong to + * {@link SearchStagesOptions.schema} (matched by `class`); the stage projects + * with the schema’s own declaration object, so `assertTypeInSchema`’s identity + * check inside {@link projectRoots} always holds. + */ + searchType: SearchType; + /** + * The selector variable that binds this type’s roots – the CONSTRUCT subject + * the batch is complete for. Must **not** be `dataset`: `?dataset` is + * substituted with the dataset IRI by the SPARQL reader, so a root bound to it + * would never reach the projection. {@link selectByClass} defaults to `root`. + */ + rootVariable: string; + /** + * Selects this type’s roots, one binding of {@link rootVariable} per root. + * Root selection is the deployment’s concern; {@link selectByClass} is a + * convenience for the object grain, not a default. + */ + itemSelector: ItemSelector; + /** Reader(s) that extract each selected root’s quads. */ + readers: StageReaders; + /** + * Roots (and so documents) per batch – the memory bound. Under a root-bound + * selector it moves memory and request count, never output. + * @default 10 + */ + batchSize?: number; + /** Maximum concurrent in-flight SPARQL queries for this stage. @default 10 */ + maxConcurrency?: number; + /** + * Capacity of the bounded queue funnelling this stage’s projected documents + * into the write. A projected document is far heavier than a quad, so lower it + * where documents are large. @default 128 + */ + queueCapacity?: number; +} + +/** Options for {@link searchStages}. */ +export interface SearchStagesOptions { + /** + * The declarative schema driving projection: one {@link SearchType} per root + * type. Every {@link SearchStageType.searchType} must be a member of it. + */ + schema: SearchSchema; + /** One entry per root type to index, each its own stage. */ + types: readonly SearchStageType[]; +} + +/** + * Compose one projecting {@link Stage} per root type – the source side of a + * search pipeline. Each stage selects its own roots, extracts each root’s quads, + * and projects the root-complete batch into tagged {@link TypedSearchDocument}s + * ({@link projectRoots} + the `searchType` tag), which the pipeline’s single + * {@link searchIndexWriter} terminal routes to that type’s collection. Projection + * happens **inside the batch**, so memory is bounded by `batchSize` roots, never + * by the dataset + * ([ADR 13](https://github.com/ldelements/lde/blob/main/docs/decisions/0013-project-inside-the-batch-per-root-type.md)). + * + * Wire the result as `new Pipeline({ datasetSelector, + * stages: searchStages(...), writers: searchIndexWriter(...) })` – one terminal, + * N stages. + */ +export function searchStages( + options: SearchStagesOptions, +): Stage[] { + const { schema, types } = options; + return types.map((type) => { + // Project with the schema’s OWN declaration object, whatever the caller + // passed: `assertTypeInSchema` (inside `projectRoots`) is an identity check, + // and re-resolving here makes a class-equal lookalike work too. + const searchType = schema.get(type.searchType.class); + if (searchType === undefined) { + throw new Error( + `Search type “${type.searchType.name}” (class ${type.searchType.class}) is not in the schema; searchStages projects only types the schema declares.`, + ); + } + const { rootVariable } = type; + return new Stage({ + name: searchType.name, + readers: type.readers, + itemSelector: type.itemSelector, + batchSize: type.batchSize, + maxConcurrency: type.maxConcurrency, + queueCapacity: type.queueCapacity, + project: async function* (quads, context) { + // The batch is root-complete by construction: `context.bindings` are the + // selector rows the readers ran with, so these are exactly this batch’s + // roots. Project them, then re-attach the type the stage was built for. + const roots = context.bindings.map( + (binding) => binding[rootVariable].value, + ); + for await (const document of projectRoots( + quads, + roots, + schema, + searchType, + )) { + yield { searchType, document }; + } + }, + }); + }); +} + +/** + * An {@link ItemSelector} that selects every instance of a root type’s source + * class: `SELECT ?‹rootVariable› WHERE { ?‹rootVariable› a }`. A + * convenience for the **object grain**, where {@link SearchType.class} really is + * the source class – **not** a default: root selection is a deployment concern, + * and three of the Dataset Register’s four catalog types have no source class at + * all (their entry point – “registered, newest registration, has a title” – is a + * deployment fact no schema states). + * + * `rootVariable` defaults to `root` and must match the stage’s + * {@link SearchStageType.rootVariable}; it must not be `dataset` (reserved by the + * SPARQL reader). + */ +export function selectByClass( + searchType: SearchType, + rootVariable = 'root', +): ItemSelector { + return new SparqlItemSelector({ + query: `SELECT ?${rootVariable} WHERE { ?${rootVariable} a <${searchType.class}> }`, + }); +} diff --git a/packages/search-pipeline/src/typed-search-document.ts b/packages/search-pipeline/src/typed-search-document.ts new file mode 100644 index 00000000..d32f7311 --- /dev/null +++ b/packages/search-pipeline/src/typed-search-document.ts @@ -0,0 +1,23 @@ +import type { SearchDocument, SearchType } from '@lde/search'; + +/** + * A projected {@link SearchDocument} tagged with the {@link SearchType} it was + * projected from. + * + * A search pipeline is N per-type stages writing to one terminal, and + * {@link https://github.com/ldelements/lde/blob/main/docs/decisions/0009-route-a-whole-schema-projection-to-per-type-collections.md | ADR 9}’s + * `searchIndexWriter` receives `write(dataset, items)` with no stage identity – + * so the type travels with the item and the writer routes each document to the + * engine run for its type by `searchType.class`. A stage mints the pair itself + * (it was constructed for one type), so nothing is ever re-derived from the + * document. + * + * It lives in `@lde/search-pipeline`, not `@lde/search`, because it exists only + * because a pipeline terminal routes: that is glue, not projection. `@lde/search` + * yields a bare {@link SearchDocument} and stays pipeline-free + * ([ADR 13](https://github.com/ldelements/lde/blob/main/docs/decisions/0013-project-inside-the-batch-per-root-type.md)). + */ +export interface TypedSearchDocument { + readonly searchType: SearchType; + readonly document: SearchDocument; +} diff --git a/packages/search-pipeline/test/multi-collection.integration.test.ts b/packages/search-pipeline/test/multi-collection.integration.test.ts index c86f4162..e3474c17 100644 --- a/packages/search-pipeline/test/multi-collection.integration.test.ts +++ b/packages/search-pipeline/test/multi-collection.integration.test.ts @@ -5,12 +5,14 @@ import type { Client } from 'typesense'; import { Dataset } from '@lde/dataset'; import type { RunContext, Writer } from '@lde/pipeline'; import { + projectRoots, searchSchema, type SearchDocument, type SearchType, } from '@lde/search'; import { BlueGreenRebuild } from '@lde/search-typesense'; import { searchIndexWriter } from '../src/search-index-writer.js'; +import type { TypedSearchDocument } from '../src/typed-search-document.js'; import { TypesenseContainer } from './typesense-container.js'; const { namedNode, literal, quad } = DataFactory; @@ -22,8 +24,8 @@ const TITLE = 'https://example.org/title'; const NAME = 'https://example.org/name'; // The Dataset Register in miniature: a `datasets` catalog collection plus one -// typed label collection (its Organization label source), built from one -// whole-schema projection. +// typed label collection (its Organization label source). Two per-type stages +// project into tagged documents; this terminal routes each to its collection. const schema = searchSchema( { name: 'Dataset', @@ -37,6 +39,9 @@ const schema = searchSchema( }, ); +const datasetType = schema.get(DATASET) as SearchType; +const organizationType = schema.get(ORGANIZATION) as SearchType; + const COLLECTION: Record = { [DATASET]: 'datasets', [ORGANIZATION]: 'organizations', @@ -61,6 +66,28 @@ function mixedQuads(): Quad[] { ]; } +/** Project `roots` of one type and tag each document, as a stage would. */ +async function tagged( + quads: readonly Quad[], + roots: readonly string[], + searchType: SearchType, +): Promise { + const documents: TypedSearchDocument[] = []; + for await (const document of projectRoots(quads, roots, schema, searchType)) { + documents.push({ searchType, document }); + } + return documents; +} + +/** The tagged documents the two stages would emit for `mixedQuads`. */ +async function mixedDocuments(): Promise { + const quads = mixedQuads(); + return [ + ...(await tagged(quads, ['https://ex/d/1'], datasetType)), + ...(await tagged(quads, ['https://ex/o/1'], organizationType)), + ]; +} + async function* stream(items: readonly Item[]): AsyncIterable { yield* items; } @@ -146,7 +173,7 @@ describe('searchIndexWriter over multiple Typesense collections', () => { writerFor: blueGreenFor(client), }).openRun(makeRunContext()); - await run.write(dataset, stream(mixedQuads())); + await run.write(dataset, stream(await mixedDocuments())); await run.flush?.(dataset, 'success'); await run.commit(); @@ -155,7 +182,7 @@ describe('searchIndexWriter over multiple Typesense collections', () => { expect(await aliasTarget(client, 'organizations')).toMatch( /^organizations_\d+$/, ); - // Each carries only the documents projected for its type. + // Each carries only the documents routed for its type. expect(await liveIds(client, 'datasets')).toEqual(['https://ex/d/1']); expect(await liveIds(client, 'organizations')).toEqual(['https://ex/o/1']); }); @@ -186,7 +213,7 @@ describe('searchIndexWriter over multiple Typesense collections', () => { const run = await searchIndexWriter({ schema, writerFor: failing }).openRun( makeRunContext(), ); - await run.write(dataset, stream(mixedQuads())); + await run.write(dataset, stream(await mixedDocuments())); await run.flush?.(dataset, 'success'); const commit = run.commit(); @@ -209,15 +236,12 @@ describe('searchIndexWriter over multiple Typesense collections', () => { }); it('commits an empty collection for a type absent from the projection, leaving the others intact', async () => { - // Only Dataset quads: the Organization projection is empty this run. - const datasetOnly = [ - quad( - namedNode('https://ex/d/1'), - namedNode(RDF_TYPE), - namedNode(DATASET), - ), - quad(namedNode('https://ex/d/1'), namedNode(TITLE), literal('Verhaal')), - ]; + // Only Dataset documents: the Organization projection is empty this run. + const datasetOnly = await tagged( + mixedQuads(), + ['https://ex/d/1'], + datasetType, + ); const run = await searchIndexWriter({ schema, @@ -227,7 +251,7 @@ describe('searchIndexWriter over multiple Typesense collections', () => { await run.flush?.(dataset, 'success'); await run.commit(); - // Datasets went live with its document; organizations went live empty — + // Datasets went live with its document; organizations went live empty – // the empty projection wiped only its own collection, never datasets. expect(await liveIds(client, 'datasets')).toEqual(['https://ex/d/1']); expect(await aliasTarget(client, 'organizations')).toMatch( @@ -241,7 +265,7 @@ describe('searchIndexWriter over multiple Typesense collections', () => { schema, writerFor: blueGreenFor(client), }).openRun(makeRunContext()); - await run.write(dataset, stream(mixedQuads())); + await run.write(dataset, stream(await mixedDocuments())); await run.flush?.(dataset, 'success'); await run.abort(new Error('run failed before commit')); diff --git a/packages/search-pipeline/test/search-index-writer.test.ts b/packages/search-pipeline/test/search-index-writer.test.ts index 487fe4b9..674f9612 100644 --- a/packages/search-pipeline/test/search-index-writer.test.ts +++ b/packages/search-pipeline/test/search-index-writer.test.ts @@ -1,6 +1,4 @@ import { describe, expect, it, vi, type Mock } from 'vitest'; -import { DataFactory } from 'n3'; -import type { Quad } from '@rdfjs/types'; import { Dataset } from '@lde/dataset'; import type { RunContext, RunWriter, Writer } from '@lde/pipeline'; import { @@ -9,37 +7,49 @@ import { type SearchType, } from '@lde/search'; import { searchIndexWriter } from '../src/search-index-writer.js'; +import type { TypedSearchDocument } from '../src/typed-search-document.js'; -const { namedNode, literal, quad } = DataFactory; - -const RDF_TYPE = 'http://www.w3.org/1999/02/22-rdf-syntax-ns#type'; const PERSON = 'https://example.org/Person'; const WORK = 'https://example.org/CreativeWork'; -const NAME = 'https://example.org/name'; const schema = searchSchema( { name: 'person', class: PERSON, - fields: [{ name: 'name', kind: 'keyword', path: NAME }], + fields: [ + { name: 'name', kind: 'keyword', path: 'https://example.org/name' }, + ], }, { name: 'work', class: WORK, - fields: [{ name: 'name', kind: 'keyword', path: NAME }], + fields: [ + { name: 'name', kind: 'keyword', path: 'https://example.org/name' }, + ], }, ); +/** The schema’s own declaration object for a class (identity matters). */ +function typeOf(classIri: string): SearchType { + const searchType = schema.get(classIri); + if (searchType === undefined) { + throw new Error(`no such type: ${classIri}`); + } + return searchType; +} + const dataset = new Dataset({ iri: new URL('http://example.org/dataset/1'), distributions: [], }); -function typedQuads(iri: string, type: string, name: string): Quad[] { - return [ - quad(namedNode(iri), namedNode(RDF_TYPE), namedNode(type)), - quad(namedNode(iri), namedNode(NAME), literal(name)), - ]; +/** A tagged document, as a per-type stage would emit it. */ +function typed( + classIri: string, + id: string, + name: string, +): TypedSearchDocument { + return { searchType: typeOf(classIri), document: { id, name: [name] } }; } async function* stream(items: readonly Item[]): AsyncIterable { @@ -75,6 +85,7 @@ function makeFleet( commit?: (searchType: SearchType) => Promise; flush?: (searchType: SearchType) => Promise; reset?: (searchType: SearchType) => Promise; + write?: (searchType: SearchType) => Promise; } = {}, ) { const collections = new Map(); @@ -103,6 +114,7 @@ function makeFleet( collected.push(document); } collection.writes.push({ dataset: written, documents: collected }); + await overrides.write?.(searchType); }, // Record the call first, then run any injected failure, so a test can // assert every collection was reached even when one of them throws. @@ -131,7 +143,7 @@ function makeFleet( function openRun( fleet: ReturnType, -): Promise> { +): Promise> { return searchIndexWriter({ schema, writerFor: fleet.writerFor }).openRun( makeRunContext(), ); @@ -149,27 +161,25 @@ const ids = (documents: SearchDocument[]) => documents.map((document) => document.id); describe('searchIndexWriter', () => { - it('routes each type’s documents to its own collection on flush', async () => { + it('routes each type’s documents to its own collection', async () => { const fleet = makeFleet(); const run = await openRun(fleet); await run.write( dataset, stream([ - ...typedQuads('http://example.org/person/1', PERSON, 'Alice'), - ...typedQuads('http://example.org/work/1', WORK, 'Nachtwacht'), + typed(PERSON, 'http://example.org/person/1', 'Alice'), + typed(WORK, 'http://example.org/work/1', 'Nachtwacht'), ]), ); await run.flush?.(dataset, 'success'); const person = collectionOf(fleet, PERSON); const work = collectionOf(fleet, WORK); - expect(person.writes).toHaveLength(1); - expect(ids(person.writes[0].documents)).toEqual([ + expect(ids(person.writes.flatMap((write) => write.documents))).toEqual([ 'http://example.org/person/1', ]); - expect(work.writes).toHaveLength(1); - expect(ids(work.writes[0].documents)).toEqual([ + expect(ids(work.writes.flatMap((write) => write.documents))).toEqual([ 'http://example.org/work/1', ]); }); @@ -184,21 +194,21 @@ describe('searchIndexWriter', () => { } }); - it('combines multiple writes for one dataset into a single projection', async () => { + it('streams every document of one write straight to its run', async () => { const fleet = makeFleet(); const run = await openRun(fleet); await run.write( dataset, - stream(typedQuads('http://example.org/person/1', PERSON, 'Alice')), - ); - await run.write( - dataset, - stream(typedQuads('http://example.org/person/2', PERSON, 'Bob')), + stream([ + typed(PERSON, 'http://example.org/person/1', 'Alice'), + typed(PERSON, 'http://example.org/person/2', 'Bob'), + ]), ); await run.flush?.(dataset, 'success'); const person = collectionOf(fleet, PERSON); + // One write call, one lane, both documents delivered in order. expect(person.writes).toHaveLength(1); expect(ids(person.writes[0].documents)).toEqual([ 'http://example.org/person/1', @@ -206,18 +216,17 @@ describe('searchIndexWriter', () => { ]); }); - it('does not accumulate quads across dataset flushes', async () => { + it('delivers documents from separate write calls independently', async () => { const fleet = makeFleet(); const run = await openRun(fleet); await run.write( dataset, - stream(typedQuads('http://example.org/person/1', PERSON, 'Alice')), + stream([typed(PERSON, 'http://example.org/person/1', 'Alice')]), ); - await run.flush?.(dataset, 'success'); await run.write( dataset, - stream(typedQuads('http://example.org/person/2', PERSON, 'Bob')), + stream([typed(PERSON, 'http://example.org/person/2', 'Bob')]), ); await run.flush?.(dataset, 'success'); @@ -232,10 +241,10 @@ describe('searchIndexWriter', () => { const fleet = makeFleet(); const run = await openRun(fleet); - // Only Person quads: the Work collection receives no documents this dataset. + // Only Person documents: the Work collection receives none this dataset. await run.write( dataset, - stream(typedQuads('http://example.org/person/1', PERSON, 'Alice')), + stream([typed(PERSON, 'http://example.org/person/1', 'Alice')]), ); await run.flush?.(dataset, 'success'); @@ -257,11 +266,11 @@ describe('searchIndexWriter', () => { } }); - it('writes no documents for a dataset whose extraction was empty, but still flushes', async () => { + it('writes nothing for an empty document stream, but still flushes', async () => { const fleet = makeFleet(); const run = await openRun(fleet); - await run.write(dataset, stream([])); + await run.write(dataset, stream([])); await run.flush?.(dataset, 'success'); for (const collection of fleet.collections.values()) { @@ -270,9 +279,42 @@ describe('searchIndexWriter', () => { } }); + it('rejects a document whose type is not in the writer’s schema', async () => { + const fleet = makeFleet(); + const run = await openRun(fleet); + const foreign: TypedSearchDocument = { + searchType: { + name: 'Ghost', + class: 'https://example.org/Ghost', + fields: [], + }, + document: { id: 'x' }, + }; + + await expect(run.write(dataset, stream([foreign]))).rejects.toThrow( + /not in this writer’s schema/, + ); + }); + + it('surfaces a run’s write failure rather than swallowing it', async () => { + const failure = new Error('import failed'); + const fleet = makeFleet({ + write: (searchType) => + searchType.class === PERSON + ? Promise.reject(failure) + : Promise.resolve(), + }); + const run = await openRun(fleet); + + await expect( + run.write( + dataset, + stream([typed(PERSON, 'http://example.org/person/1', 'Alice')]), + ), + ).rejects.toBe(failure); + }); + it('flushes every collection even when one collection’s flush fails, then surfaces it', async () => { - // The Person collection's flush (e.g. a rollback of a failed dataset) - // throws; the Work collection must still be flushed, not skipped. const failure = new Error('person rollback failed'); const fleet = makeFleet({ flush: (searchType) => @@ -310,49 +352,31 @@ describe('searchIndexWriter', () => { expect(collectionOf(fleet, PERSON).resets).toEqual([dataset]); }); - it('reset discards the buffered pass and forwards to every collection', async () => { + it('reset forwards to every collection', async () => { const fleet = makeFleet(); const run = await openRun(fleet); - await run.write( - dataset, - stream(typedQuads('http://example.org/person/1', PERSON, 'Endpoint')), - ); await run.reset?.(dataset); - await run.write( - dataset, - stream(typedQuads('http://example.org/person/1', PERSON, 'Dump')), - ); - await run.flush?.(dataset, 'success'); - const person = collectionOf(fleet, PERSON); - expect(person.resets).toEqual([dataset]); + expect(collectionOf(fleet, PERSON).resets).toEqual([dataset]); expect(collectionOf(fleet, WORK).resets).toEqual([dataset]); - expect(person.writes).toHaveLength(1); - expect(person.writes[0].documents).toEqual([ - { id: 'http://example.org/person/1', name: ['Dump'] }, - ]); }); - it('projects never-flushed leftovers on commit, then commits every collection', async () => { + it('commits every collection', async () => { const fleet = makeFleet(); const run = await openRun(fleet); await run.write( dataset, - stream(typedQuads('http://example.org/person/1', PERSON, 'Alice')), + stream([typed(PERSON, 'http://example.org/person/1', 'Alice')]), ); await run.commit(); - const person = collectionOf(fleet, PERSON); - expect(person.writes).toHaveLength(1); - expect(person.commit).toHaveBeenCalledOnce(); + expect(collectionOf(fleet, PERSON).commit).toHaveBeenCalledOnce(); expect(collectionOf(fleet, WORK).commit).toHaveBeenCalledOnce(); }); it('commits every collection independently and aggregates the failures', async () => { - // The Work collection’s commit fails; the Person collection must still go - // live, and the failure must surface as an AggregateError. const failure = new Error('work collection swap failed'); const fleet = makeFleet({ commit: (searchType) => @@ -362,10 +386,8 @@ describe('searchIndexWriter', () => { await expect(run.commit()).rejects.toThrow(AggregateError); - const person = collectionOf(fleet, PERSON); - const work = collectionOf(fleet, WORK); - expect(person.commit).toHaveBeenCalledOnce(); - expect(work.commit).toHaveBeenCalledOnce(); + expect(collectionOf(fleet, PERSON).commit).toHaveBeenCalledOnce(); + expect(collectionOf(fleet, WORK).commit).toHaveBeenCalledOnce(); }); it('abort after a partial commit finalizes only the collections that did not go live', async () => { @@ -448,16 +470,16 @@ describe('searchIndexWriter', () => { await run.write( dataset, - stream(typedQuads('http://example.org/person/1', PERSON, 'Alice')), + stream([typed(PERSON, 'http://example.org/person/1', 'Alice')]), ); await run.reset?.(dataset); await run.write( dataset, - stream(typedQuads('http://example.org/person/1', PERSON, 'Alice')), + stream([typed(PERSON, 'http://example.org/person/1', 'Alice')]), ); await run.flush?.(dataset, 'success'); - expect(written).toHaveLength(1); + expect(written).toHaveLength(2); }); it('opens every collection’s run with the pipeline’s run context', async () => { diff --git a/packages/search-pipeline/test/search-stages.test.ts b/packages/search-pipeline/test/search-stages.test.ts new file mode 100644 index 00000000..47c00cdc --- /dev/null +++ b/packages/search-pipeline/test/search-stages.test.ts @@ -0,0 +1,261 @@ +import { describe, expect, it } from 'vitest'; +import { DataFactory } from 'n3'; +import type { Quad } from '@rdfjs/types'; +import { Dataset, Distribution } from '@lde/dataset'; +import { + SparqlItemSelector, + Stage, + type DatasetWriter, + type ItemSelector, + type Reader, + type VariableBindings, +} from '@lde/pipeline'; +import { projectRoots, searchSchema, type SearchType } from '@lde/search'; +import { searchStages, selectByClass } from '../src/search-stages.js'; +import type { TypedSearchDocument } from '../src/typed-search-document.js'; + +const { namedNode, literal, quad } = DataFactory; + +const PERSON = 'https://example.org/Person'; +const NAME = 'https://example.org/name'; + +const schema = searchSchema({ + name: 'Person', + class: PERSON, + fields: [{ name: 'name', kind: 'keyword', path: NAME }], +}); +const person = schema.get(PERSON) as SearchType; + +const dataset = new Dataset({ + iri: new URL('http://example.org/dataset/1'), + distributions: [], +}); +const distribution = new Distribution(new URL('http://example.org/sparql')); + +async function* stream(items: readonly Item[]): AsyncIterable { + yield* items; +} + +/** A selector that yields one `?root` binding per given IRI. */ +function rootsSelector(roots: readonly string[]): ItemSelector { + return { + select: async function* (): AsyncIterable { + for (const iri of roots) { + yield { root: namedNode(iri) }; + } + }, + }; +} + +/** A reader that emits one `name` triple per root in the batch’s bindings. */ +const nameReader: Reader = { + read: (_dataset, _distribution, options) => { + const quads: Quad[] = []; + for (const binding of options?.bindings ?? []) { + const root = binding.root.value; + quads.push( + quad(namedNode(root), namedNode(NAME), literal(`Name ${root}`)), + ); + } + return Promise.resolve(stream(quads)); + }, +}; + +describe('searchStages', () => { + it('projects each root type over its selector’s roots, tagged with its type', async () => { + const [stage] = searchStages({ + schema, + types: [ + { + searchType: person, + rootVariable: 'root', + itemSelector: rootsSelector(['https://ex/p/1', 'https://ex/p/2']), + readers: nameReader, + }, + ], + }); + + const received: TypedSearchDocument[] = []; + const writer: DatasetWriter = { + write: async (_dataset, items) => { + for await (const item of items) { + received.push(item); + } + }, + }; + + await stage.run(dataset, distribution, writer); + + expect(received.map((item) => item.searchType)).toEqual([person, person]); + expect(received.map((item) => item.document)).toEqual([ + { id: 'https://ex/p/1', name: ['Name https://ex/p/1'] }, + { id: 'https://ex/p/2', name: ['Name https://ex/p/2'] }, + ]); + }); + + it('names each stage after its type and yields one stage per type', () => { + const stages = searchStages({ + schema, + types: [ + { + searchType: person, + rootVariable: 'root', + itemSelector: rootsSelector([]), + readers: nameReader, + }, + ], + }); + expect(stages).toHaveLength(1); + expect(stages[0].name).toBe('Person'); + }); + + it('projects with the schema’s own declaration even when handed a class-equal lookalike', async () => { + // A reconstructed object with the same class – `assertTypeInSchema` is an + // identity check, so searchStages must re-resolve to the schema’s own object. + const lookalike: SearchType = { + name: 'Person', + class: PERSON, + fields: [{ name: 'name', kind: 'keyword', path: NAME }], + }; + const [stage] = searchStages({ + schema, + types: [ + { + searchType: lookalike, + rootVariable: 'root', + itemSelector: rootsSelector(['https://ex/p/1']), + readers: nameReader, + }, + ], + }); + + const received: TypedSearchDocument[] = []; + await stage.run(dataset, distribution, { + write: async (_dataset, items) => { + for await (const item of items) received.push(item); + }, + }); + + expect(received).toHaveLength(1); + // Tagged with the schema’s own object, not the lookalike. + expect(received[0].searchType).toBe(person); + }); + + it('throws when a type is not in the schema', () => { + const foreign: SearchType = { + name: 'Ghost', + class: 'https://example.org/Ghost', + fields: [], + }; + expect(() => + searchStages({ + schema, + types: [ + { + searchType: foreign, + rootVariable: 'root', + itemSelector: rootsSelector([]), + readers: nameReader, + }, + ], + }), + ).toThrow(/not in the schema/); + }); +}); + +describe('selectByClass', () => { + it('builds a SPARQL selector for the type’s source class', () => { + const selector = selectByClass(person); + expect(selector).toBeInstanceOf(SparqlItemSelector); + }); + + it('accepts a custom root variable', () => { + expect(() => selectByClass(person, 'subject')).not.toThrow(); + }); +}); + +describe('memory is bounded by batchSize, not the input (counting, not measuring)', () => { + const BATCH_SIZE = 10; + + /** + * Run one projecting stage over `rootCount` synthetic roots and record what + * the batch structures actually held. The `project` closure mirrors what + * {@link searchStages} builds, wrapped to count invocations and gauge live + * documents; `maxConcurrency` and `queueCapacity` are pinned to 1 so the + * per-batch peak is deterministic and independent of the batch count. + */ + async function runOver(rootCount: number) { + const roots = Array.from( + { length: rootCount }, + (_, index) => `https://ex/p/${index}`, + ); + + const quadsPerProject: number[] = []; + let liveDocuments = 0; + let peakLiveDocuments = 0; + + const stage = new Stage({ + name: 'Person', + readers: nameReader, + itemSelector: rootsSelector(roots), + batchSize: BATCH_SIZE, + maxConcurrency: 1, + queueCapacity: 1, + project: async function* (quads, context) { + quadsPerProject.push(quads.length); + const batchRoots = context.bindings.map( + (binding) => binding.root.value, + ); + for await (const document of projectRoots( + quads, + batchRoots, + schema, + person, + )) { + liveDocuments += 1; + peakLiveDocuments = Math.max(peakLiveDocuments, liveDocuments); + yield { searchType: person, document }; + } + }, + }); + + const writer: DatasetWriter = { + write: async (_dataset, items) => { + for await (const _item of items) { + // A document reaches the terminal and is released: no longer live. + liveDocuments -= 1; + } + }, + }; + + await stage.run(dataset, distribution, writer); + + return { + projectInvocations: quadsPerProject.length, + maxQuadsPerProject: Math.max(...quadsPerProject), + peakLiveDocuments, + }; + } + + it('projects once per batch and never grows a structure with the input', async () => { + const small = await runOver(10); + const large = await runOver(10_000); + + // (a) project is invoked exactly ceil(roots / batchSize) times. + expect(small.projectInvocations).toBe(1); + expect(large.projectInvocations).toBe(1000); + + // (b) the largest batch handed to project is one batchSize either way – one + // name quad per root, so batchSize quads – never the whole input. + expect(small.maxQuadsPerProject).toBe(BATCH_SIZE); + expect(large.maxQuadsPerProject).toBe(BATCH_SIZE); + + // (c) the peak number of documents alive between projection and the writer + // is identical at both input sizes – the whole point of #606. A buffering + // writer’s peak would be every document (10 vs 10 000), so this equality + // fails against it and holds here. + expect(large.peakLiveDocuments).toBe(small.peakLiveDocuments); + // …and it is a small constant, nowhere near the 10 000 roots. + expect(large.peakLiveDocuments).toBeLessThanOrEqual(BATCH_SIZE); + }); +}); From 7d0925f34de9bd3167f434d015c6c6ed37c17b45 Mon Sep 17 00:00:00 2001 From: David de Boer Date: Sat, 18 Jul 2026 19:57:42 +0200 Subject: [PATCH 3/4] fix(search-pipeline): fail clearly when a selector omits the stage's rootVariable The batch project closure dereferenced binding[rootVariable].value directly, so a config mismatch (the stage's rootVariable differs from the selector's projected variable) threw an opaque TypeError. Guard the deref and throw a named error that points at the type and the unbound variable. --- packages/search-pipeline/src/search-stages.ts | 18 +++++++--- .../test/search-stages.test.ts | 33 +++++++++++++++++-- 2 files changed, 44 insertions(+), 7 deletions(-) diff --git a/packages/search-pipeline/src/search-stages.ts b/packages/search-pipeline/src/search-stages.ts index 82cb905b..895a7af2 100644 --- a/packages/search-pipeline/src/search-stages.ts +++ b/packages/search-pipeline/src/search-stages.ts @@ -61,8 +61,8 @@ export interface SearchStagesOptions { /** * Compose one projecting {@link Stage} per root type – the source side of a * search pipeline. Each stage selects its own roots, extracts each root’s quads, - * and projects the root-complete batch into tagged {@link TypedSearchDocument}s - * ({@link projectRoots} + the `searchType` tag), which the pipeline’s single + * and projects the root-complete batch into {@link TypedSearchDocument}s + * ({@link projectRoots} + the `searchType` pair), which the pipeline’s single * {@link searchIndexWriter} terminal routes to that type’s collection. Projection * happens **inside the batch**, so memory is bounded by `batchSize` roots, never * by the dataset @@ -98,9 +98,17 @@ export function searchStages( // The batch is root-complete by construction: `context.bindings` are the // selector rows the readers ran with, so these are exactly this batch’s // roots. Project them, then re-attach the type the stage was built for. - const roots = context.bindings.map( - (binding) => binding[rootVariable].value, - ); + const roots = context.bindings.map((binding) => { + const term = binding[rootVariable]; + if (term === undefined) { + // The selector projected a different variable than the stage reads: + // a config mismatch. Fail loudly rather than deref `undefined`. + throw new Error( + `Stage “${searchType.name}”: selector did not bind ?${rootVariable} – the stage’s rootVariable must match the selector’s projected variable.`, + ); + } + return term.value; + }); for await (const document of projectRoots( quads, roots, diff --git a/packages/search-pipeline/test/search-stages.test.ts b/packages/search-pipeline/test/search-stages.test.ts index 47c00cdc..377d6f98 100644 --- a/packages/search-pipeline/test/search-stages.test.ts +++ b/packages/search-pipeline/test/search-stages.test.ts @@ -62,7 +62,7 @@ const nameReader: Reader = { }; describe('searchStages', () => { - it('projects each root type over its selector’s roots, tagged with its type', async () => { + it('projects each root type over its selector’s roots, paired with its type', async () => { const [stage] = searchStages({ schema, types: [ @@ -137,7 +137,7 @@ describe('searchStages', () => { }); expect(received).toHaveLength(1); - // Tagged with the schema’s own object, not the lookalike. + // Paired with the schema’s own object, not the lookalike. expect(received[0].searchType).toBe(person); }); @@ -161,6 +161,35 @@ describe('searchStages', () => { }), ).toThrow(/not in the schema/); }); + + it('fails clearly when the selector does not bind the stage’s rootVariable', async () => { + // The stage reads ?subject, but `rootsSelector` binds ?root – a config + // mismatch. The batch deref must throw a named error, not an opaque + // `Cannot read properties of undefined`. + const [stage] = searchStages({ + schema, + types: [ + { + searchType: person, + rootVariable: 'subject', + itemSelector: rootsSelector(['https://ex/p/1']), + readers: nameReader, + }, + ], + }); + + const writer: DatasetWriter = { + write: async (_dataset, items) => { + for await (const _item of items) { + // drain + } + }, + }; + + await expect(stage.run(dataset, distribution, writer)).rejects.toThrow( + /did not bind \?subject/, + ); + }); }); describe('selectByClass', () => { From dc44d1ad3f7d81726aba2ffea4e2503f0e65600b Mon Sep 17 00:00:00 2001 From: David de Boer Date: Sat, 18 Jul 2026 19:57:52 +0200 Subject: [PATCH 4/4] docs(search): fix the GraphQL indexing example and drop the tag metaphor The search-api-graphql README credited searchSchema with indexing (it only builds a schema, and the line was an unassigned no-op); clarify that the pipeline indexes the superset schema while the GraphQL API serves a subset. Reword the frameSubjects comment, which still referenced the removed rootType. Rename the document-SearchType relationship from tag to pair throughout (TypedSearchDocument is literally a pair); BCP-47 language-tag wording is left untouched. --- packages/search-api-graphql/README.md | 13 ++++++++----- packages/search-pipeline/README.md | 6 +++--- .../search-pipeline/src/search-index-writer.ts | 10 +++++----- .../search-pipeline/src/typed-search-document.ts | 2 +- .../test/multi-collection.integration.test.ts | 15 ++++++++------- .../test/search-index-writer.test.ts | 2 +- packages/search/README.md | 4 ++-- packages/search/src/frame-by-type.ts | 11 ++++++----- packages/search/src/project.ts | 2 +- packages/search/test/project.test.ts | 2 +- 10 files changed, 36 insertions(+), 31 deletions(-) diff --git a/packages/search-api-graphql/README.md b/packages/search-api-graphql/README.md index 069c2bad..8b192f3f 100644 --- a/packages/search-api-graphql/README.md +++ b/packages/search-api-graphql/README.md @@ -63,14 +63,17 @@ types such as a common `Agent`) are created once and reused across root types. `types` never filters: every `SearchType` in the schema you pass gets a root field (options for a type not in the schema are a build-time error). To expose -only part of what you index, narrow the **schema argument** -(`searchSchema(…)` is a cheap constructor): +only part of what you index, narrow the **schema argument** you hand +`buildGraphQLSchema` (`searchSchema(…)` is a cheap constructor, so build one per +consumer): ```ts -// Index all three types (a projecting stage per type, see @lde/search-pipeline)… -searchSchema(DATASET, PERSON, INTERNAL); +// Index a superset: hand a three-type schema to the pipeline, which projects and +// stores one collection per type (see @lde/search-pipeline). INTERNAL is indexed +// (e.g. a label source references resolve against) but never served. +const indexed = searchSchema(DATASET, PERSON, INTERNAL); -// …but serve only two. +// Serve a subset: the GraphQL API exposes only two of those types. const gqlSchema = buildGraphQLSchema(searchSchema(DATASET, PERSON)); ``` diff --git a/packages/search-pipeline/README.md b/packages/search-pipeline/README.md index 7d4941c4..2ad6b74d 100644 --- a/packages/search-pipeline/README.md +++ b/packages/search-pipeline/README.md @@ -18,12 +18,12 @@ The division of labour ([ADR 6](../../docs/decisions/0006-make-the-writer-transa - **this package** composes them: - `searchStages` builds one projecting `Stage` per root type. Each stage selects its own roots, extracts each root’s quads, and projects the - **root-complete batch** into documents tagged with their `SearchType` + **root-complete batch** into documents paired with their `SearchType` (`TypedSearchDocument`) – so projection happens inside the batch and memory is bounded by `batchSize` roots, not the dataset ([ADR 13](../../docs/decisions/0013-project-inside-the-batch-per-root-type.md)); - `searchIndexWriter` is the pipeline’s single terminal: an engine-agnostic - router that dispatches each tagged document to the engine writer for **its** + router that dispatches each document to the engine writer for **its** type’s collection ([ADR 9](../../docs/decisions/0009-route-a-whole-schema-projection-to-per-type-collections.md)). It owns no projection and buffers nothing. @@ -136,7 +136,7 @@ given the same names (`collections`), or it reads the derived ones and finds an empty index. Each stage projects its own root-complete batches (`@lde/search`’s -`projectRoots`) into documents tagged with its `SearchType`, and the terminal +`projectRoots`) into documents paired with its `SearchType`, and the terminal dispatches each to the engine run for its type’s collection as it arrives – before the engine acts on the dataset’s completion, so an In-place stale sweep never races its own documents. The run lifecycle (run context, per-dataset flush diff --git a/packages/search-pipeline/src/search-index-writer.ts b/packages/search-pipeline/src/search-index-writer.ts index 3fd295e8..cb6f42a7 100644 --- a/packages/search-pipeline/src/search-index-writer.ts +++ b/packages/search-pipeline/src/search-index-writer.ts @@ -37,7 +37,7 @@ export interface SearchIndexWriterOptions { * The single terminal of a search-indexing pipeline: an engine-agnostic router * that fans already-projected documents out across a type’s collections. The * per-type {@link https://github.com/ldelements/lde/blob/main/docs/decisions/0013-project-inside-the-batch-per-root-type.md | stages} - * project inside the batch and tag each document with its {@link SearchType} + * project inside the batch and pair each document with its {@link SearchType} * ({@link TypedSearchDocument}); this writer dispatches each document to the * engine run for **its** type by `searchType.class`. It **owns no projection** * (that moved into the stages) and **buffers nothing** (documents stream through @@ -69,7 +69,7 @@ export function searchIndexWriter( ): Writer { const { schema, writerFor } = options; // One engine writer per root type, built once; each run opens them all. Keyed - // by the type IRI, which is also how a tagged document names its type. + // by the type IRI, which is also how a paired document names its type. const writers = new Map>( [...schema.values()].map((searchType) => [ searchType.class, @@ -104,9 +104,9 @@ export function searchIndexWriter( dataset: Dataset, items: AsyncIterable, ) => { - // Route each tagged document to its type’s run, streaming. A stage - // writes one type, but the terminal carries no stage identity, so it - // routes per item by the tag. Each type gets one `run.write`, fed a + // Route each document to its type’s run, streaming. A stage writes + // one type, but the terminal carries no stage identity, so it routes + // per item by its `searchType`. Each type gets one `run.write`, fed a // bounded queue the run drains concurrently – so memory stays O(batch) // per type, never O(dataset). Almost always one lane per call. const lanes = new Map< diff --git a/packages/search-pipeline/src/typed-search-document.ts b/packages/search-pipeline/src/typed-search-document.ts index d32f7311..6d92f399 100644 --- a/packages/search-pipeline/src/typed-search-document.ts +++ b/packages/search-pipeline/src/typed-search-document.ts @@ -1,7 +1,7 @@ import type { SearchDocument, SearchType } from '@lde/search'; /** - * A projected {@link SearchDocument} tagged with the {@link SearchType} it was + * A projected {@link SearchDocument} paired with the {@link SearchType} it was * projected from. * * A search pipeline is N per-type stages writing to one terminal, and diff --git a/packages/search-pipeline/test/multi-collection.integration.test.ts b/packages/search-pipeline/test/multi-collection.integration.test.ts index e3474c17..05c7a2d9 100644 --- a/packages/search-pipeline/test/multi-collection.integration.test.ts +++ b/packages/search-pipeline/test/multi-collection.integration.test.ts @@ -25,7 +25,8 @@ const NAME = 'https://example.org/name'; // The Dataset Register in miniature: a `datasets` catalog collection plus one // typed label collection (its Organization label source). Two per-type stages -// project into tagged documents; this terminal routes each to its collection. +// project into documents paired with their type; this terminal routes each to +// its collection. const schema = searchSchema( { name: 'Dataset', @@ -66,8 +67,8 @@ function mixedQuads(): Quad[] { ]; } -/** Project `roots` of one type and tag each document, as a stage would. */ -async function tagged( +/** Project `roots` of one type and pair each document with it, as a stage would. */ +async function paired( quads: readonly Quad[], roots: readonly string[], searchType: SearchType, @@ -79,12 +80,12 @@ async function tagged( return documents; } -/** The tagged documents the two stages would emit for `mixedQuads`. */ +/** The documents (paired with their type) the two stages would emit for `mixedQuads`. */ async function mixedDocuments(): Promise { const quads = mixedQuads(); return [ - ...(await tagged(quads, ['https://ex/d/1'], datasetType)), - ...(await tagged(quads, ['https://ex/o/1'], organizationType)), + ...(await paired(quads, ['https://ex/d/1'], datasetType)), + ...(await paired(quads, ['https://ex/o/1'], organizationType)), ]; } @@ -237,7 +238,7 @@ describe('searchIndexWriter over multiple Typesense collections', () => { it('commits an empty collection for a type absent from the projection, leaving the others intact', async () => { // Only Dataset documents: the Organization projection is empty this run. - const datasetOnly = await tagged( + const datasetOnly = await paired( mixedQuads(), ['https://ex/d/1'], datasetType, diff --git a/packages/search-pipeline/test/search-index-writer.test.ts b/packages/search-pipeline/test/search-index-writer.test.ts index 674f9612..ad76f890 100644 --- a/packages/search-pipeline/test/search-index-writer.test.ts +++ b/packages/search-pipeline/test/search-index-writer.test.ts @@ -43,7 +43,7 @@ const dataset = new Dataset({ distributions: [], }); -/** A tagged document, as a per-type stage would emit it. */ +/** A document paired with its type, as a per-type stage would emit it. */ function typed( classIri: string, id: string, diff --git a/packages/search/README.md b/packages/search/README.md index 43e691d9..0f9208eb 100644 --- a/packages/search/README.md +++ b/packages/search/README.md @@ -154,7 +154,7 @@ const DATASET = defineSearchType({ const schema = searchSchema(DATASET); for await (const document of projectRoots(quads, roots, schema, DATASET)) { // one flat search document per given root subject, streamed. The caller (the - // pipeline selector) supplies `roots`; tagging a document with its type for a + // pipeline selector) supplies `roots`; pairing a document with its type for a // multi-collection writer is the pipeline glue’s job (see // `@lde/search-pipeline`), not the projection’s. } @@ -230,7 +230,7 @@ predicate IRIs with language tags preserved. `assertTypeInSchema` guards that the passed `SearchType` is a member of the schema – the port’s own membership check – so no schema is ever forged to scope a projection to one type. -`projectRoots` yields a **bare** `SearchDocument`. Tagging each document with the +`projectRoots` yields a **bare** `SearchDocument`. Pairing each document with the `SearchType` it belongs to, so the write side can fan a mixed stream out to per-type collections, is a routing concern owned by the pipeline glue – see `@lde/search-pipeline`’s `searchStages` and multi-collection writer – not the diff --git a/packages/search/src/frame-by-type.ts b/packages/search/src/frame-by-type.ts index abca6ee6..ad1fde40 100644 --- a/packages/search/src/frame-by-type.ts +++ b/packages/search/src/frame-by-type.ts @@ -78,11 +78,12 @@ export async function* frameSubjects( .flatMap((quad) => bySubject.get(quad.object.value) ?? []); const subgraph = [...owned, ...referenced]; const expanded = await jsonld.fromRDF(subgraph); - // Frame for THIS specific root subject by `@id`, not just by root type. A - // one-hop reference can itself be of `rootType` (e.g. a terminology source - // that is also a separately registered dataset), so framing by type alone - // returns several root nodes and `[0]` could be the referenced one – which - // would emit it twice and drop this subject entirely. + // Frame for THIS specific root subject by its `@id`. The subgraph embeds the + // root’s one-hop references, and such a referent can itself be one of the + // requested `roots` (e.g. a terminology source that is also a separately + // registered dataset). Framing by `{ '@id': rootIri }` pins the output to + // this subject, so the referent is framed once in its own turn rather than + // surfacing here as a second `@graph` node that `[0]` might pick instead. const framed = await jsonld.frame( expanded, { '@id': rootIri }, diff --git a/packages/search/src/project.ts b/packages/search/src/project.ts index 62aba77b..b9883ad5 100644 --- a/packages/search/src/project.ts +++ b/packages/search/src/project.ts @@ -53,7 +53,7 @@ export function projectDocument( * and the projection frames each distinct subject once. {@link assertTypeInSchema} * guards that `searchType` belongs to `schema` – the port’s own membership check * – so no schema is ever forged to scope a projection to one type. Yields a bare - * {@link SearchDocument}: tagging a document with its type is a routing concern, + * {@link SearchDocument}: pairing a document with its type is a routing concern, * owned by the pipeline glue, not the projection. * * Consumes `quads` once, so it accepts any `Iterable` – a batch’s materialized diff --git a/packages/search/test/project.test.ts b/packages/search/test/project.test.ts index 96f9a498..f0a09ef0 100644 --- a/packages/search/test/project.test.ts +++ b/packages/search/test/project.test.ts @@ -580,7 +580,7 @@ describe('projectRoots', () => { ]); }); - it('yields a bare document (no searchType tag)', async () => { + it('yields a bare document, not paired with a searchType', async () => { const quads = new Parser({ format: 'N-Triples' }).parse( ` <${dcterms.title.value}> "Titel"@nl .`, );