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..8b192f3f 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 @@ -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… -projectGraph(quads, 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)); ``` @@ -98,7 +101,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 +117,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..2ad6b74d 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 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 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 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 +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..cb6f42a7 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 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 + * 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 paired 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 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< + 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..895a7af2 --- /dev/null +++ b/packages/search-pipeline/src/search-stages.ts @@ -0,0 +1,145 @@ +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 {@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 + * ([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) => { + 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, + 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..6d92f399 --- /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} paired 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..05c7a2d9 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,9 @@ 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 documents paired with their type; this terminal routes each to +// its collection. const schema = searchSchema( { name: 'Dataset', @@ -37,6 +40,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 +67,28 @@ function mixedQuads(): Quad[] { ]; } +/** 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, +): Promise { + const documents: TypedSearchDocument[] = []; + for await (const document of projectRoots(quads, roots, schema, searchType)) { + documents.push({ searchType, document }); + } + return documents; +} + +/** The documents (paired with their type) the two stages would emit for `mixedQuads`. */ +async function mixedDocuments(): Promise { + const quads = mixedQuads(); + return [ + ...(await paired(quads, ['https://ex/d/1'], datasetType)), + ...(await paired(quads, ['https://ex/o/1'], organizationType)), + ]; +} + async function* stream(items: readonly Item[]): AsyncIterable { yield* items; } @@ -146,7 +174,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 +183,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 +214,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 +237,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 paired( + mixedQuads(), + ['https://ex/d/1'], + datasetType, + ); const run = await searchIndexWriter({ schema, @@ -227,7 +252,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 +266,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..ad76f890 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 document paired with its type, 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..377d6f98 --- /dev/null +++ b/packages/search-pipeline/test/search-stages.test.ts @@ -0,0 +1,290 @@ +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, paired 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); + // Paired 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/); + }); + + 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', () => { + 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); + }); +}); diff --git a/packages/search/README.md b/packages/search/README.md index 34f85a6f..0f9208eb 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`; 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. } ``` @@ -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`. 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 +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..ad1fde40 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, @@ -98,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 }, @@ -114,18 +95,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..b9883ad5 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}: 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 * 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..f0a09ef0 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); @@ -647,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 .`, ); 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, }, },