diff --git a/docs/reference/search-indexer.md b/docs/reference/search-indexer.md index e4bf443f..a28b2ba8 100644 --- a/docs/reference/search-indexer.md +++ b/docs/reference/search-indexer.md @@ -64,24 +64,25 @@ images. ## Configuration -| Variable | Default | Meaning | -| -------------------- | --------------------------- | ----------------------------------------------------------------------------------------------------------------------------------------------------- | -| `SCHEMA_MODULE` | `/config/search-schema.mjs` | Path of the mounted schema-declaration module | -| `REGISTRY_ENDPOINT` | **required** | SPARQL endpoint of the DCAT dataset registry | -| `DATASETS` | all registry datasets | Dataset IRIs to index (whitespace- or comma-separated) | -| `DATASET_CRITERIA` | all registry datasets | Search criteria as a JSON object, in [`@lde/dataset-registry-client`](./dataset-registry-client)’s criteria format (mutually exclusive w/ `DATASETS`) | -| `TYPESENSE_HOST` | **required** | Typesense host | -| `TYPESENSE_PORT` | `8108` | Typesense port | -| `TYPESENSE_PROTOCOL` | `http` | `http` or `https` | -| `TYPESENSE_API_KEY` | **required** | An admin key: the indexer creates, writes and swaps collections | -| `REBUILD_MODE` | `in-place` | `in-place` (update the live collection) or `blue-green` (swap on commit) | -| `COLLECTION_PREFIX` | none | Prefix for every derived collection name (configure the read side to match) | -| `PROVENANCE_FILE` | none | JSON file remembering per-dataset processing, to skip unchanged datasets | -| `PIPELINE_VERSION` | none | Version keying the skip decisions; required with `PROVENANCE_FILE` | -| `QLEVER_IMAGE` | none | Enables the QLever import path (see below), e.g. `adfreiburg/qlever:latest` | -| `IMPORT_STRATEGY` | `sparql` | `sparql`, `sparqlWithImportFallback` or `import`; requires `QLEVER_IMAGE` | -| `DATA_DIR` | `/data` | Directory for downloaded dumps and QLever index caches | -| `QLEVER_NETWORK` | none | Docker network the spawned QLever joins; set when the indexer itself runs containerized (see below) | +| Variable | Default | Meaning | +| --------------------- | --------------------------- | ----------------------------------------------------------------------------------------------------------------------------------------------------- | +| `SCHEMA_MODULE` | `/config/search-schema.mjs` | Path of the mounted schema-declaration module | +| `REGISTRY_ENDPOINT` | **required** | SPARQL endpoint of the DCAT dataset registry | +| `REGISTRY_ROOT_TYPES` | none | Root types extracted from the registry instead of the dataset’s distribution (see below), by schema type name | +| `DATASETS` | all registry datasets | Dataset IRIs to index (whitespace- or comma-separated) | +| `DATASET_CRITERIA` | all registry datasets | Search criteria as a JSON object, in [`@lde/dataset-registry-client`](./dataset-registry-client)’s criteria format (mutually exclusive w/ `DATASETS`) | +| `TYPESENSE_HOST` | **required** | Typesense host | +| `TYPESENSE_PORT` | `8108` | Typesense port | +| `TYPESENSE_PROTOCOL` | `http` | `http` or `https` | +| `TYPESENSE_API_KEY` | **required** | An admin key: the indexer creates, writes and swaps collections | +| `REBUILD_MODE` | `in-place` | `in-place` (update the live collection) or `blue-green` (swap on commit) | +| `COLLECTION_PREFIX` | none | Prefix for every derived collection name (configure the read side to match) | +| `PROVENANCE_FILE` | none | JSON file remembering per-dataset processing, to skip unchanged datasets | +| `PIPELINE_VERSION` | none | Version keying the skip decisions; required with `PROVENANCE_FILE` | +| `QLEVER_IMAGE` | none | Enables the QLever import path (see below), e.g. `adfreiburg/qlever:latest` | +| `IMPORT_STRATEGY` | `sparql` | `sparql`, `sparqlWithImportFallback` or `import`; requires `QLEVER_IMAGE` | +| `DATA_DIR` | `/data` | Directory for downloaded dumps and QLever index caches | +| `QLEVER_NETWORK` | none | Docker network the spawned QLever joins; set when the indexer itself runs containerized (see below) | A misconfigured boot reports **all** problems in one error, not one per crash loop. `PROVENANCE_FILE` must sit on a durable volume, and cannot be combined @@ -92,6 +93,40 @@ that ownership, so a named volume mounted there just works, while a bind mount must be `chown`ed on the host. A run fails at start when the file is not writable, instead of silently never skipping. +## Registry-sourced root types + +Every root type is extracted from the dataset’s **distribution** by default – +its live SPARQL endpoint, or a dump imported into QLever. That holds while a +type is described by the data the dataset publishes, and it breaks for the +dataset itself: a dataset’s description is governed by a different application +profile from the objects it contains, and it lives in the register. Nothing +obliges a publisher to describe its own dataset inside its dump, so a `Dataset` +type extracted from the distribution yields documents for the few publishers +that self-describe and nothing for the rest. + +`REGISTRY_ROOT_TYPES` names the root types to extract from `REGISTRY_ENDPOINT` +instead: + +```sh +--env REGISTRY_ROOT_TYPES="Dataset Publisher" +``` + +Same CONSTRUCT generator, same framing, same projection, same writers – only +the source differs. Each such stage is **scoped to the graph the dataset in +hand names**: a register holds every registration, so an unscoped stage would +re-index the whole catalogue once per dataset processed. Scoped, one pass sees +exactly one registration, and a `selectByClass` finds that dataset’s own roots +inside it – its `dcat:Dataset` node, and the `foaf:Organization` its +`dcterms:publisher` points at. This presumes the register names each +registration’s graph after the dataset IRI, which is how a DCAT register that +crawls per registration stores it. + +Routing is deployment topology, so it is configuration and not part of the +schema: a `SearchType` is defined by its `class`, never by where its triples +come from, and the same declaration serves a deployment that sources it +differently. A name that the mounted schema does not declare fails the boot, +rather than shipping an empty collection. + ## The QLever import path By default the indexer serves only datasets that publish a live SPARQL diff --git a/docs/reference/search-pipeline.md b/docs/reference/search-pipeline.md index fad21d1f..9b74d0c7 100644 --- a/docs/reference/search-pipeline.md +++ b/docs/reference/search-pipeline.md @@ -171,6 +171,8 @@ const pipeline = new Pipeline({ }, ], }), + // A stage reads the dataset’s own distribution unless `sourceFor` says + // otherwise – see “Registry-sourced root types” below. // 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 @@ -224,6 +226,74 @@ blank-node subjects (`FILTER(!isBlank(?root))`): a blank node has no stable document key, so it can never become a search document – a custom selector should exclude them too. +## Registry-sourced root types + +A stage reads the dataset’s own distribution. That holds while a root type is +described by the data the dataset publishes, and it breaks for the dataset +itself: a dataset’s description is governed by a different application profile +from the objects it contains, and it lives in the **dataset registry**. +Registering a dataset is submitting a description, so the register covers every +dataset a pipeline can select – while nothing obliges a publisher to describe +its own dataset inside its dump. + +`registrySource` points a stage at the register instead, scoped to the graph the +dataset in hand names: + +```typescript +import { + registrySource, + searchStages, + selectByClass, +} from '@lde/search-pipeline'; + +searchStages({ + schema, + types: [ + { + searchType: dataset, + rootVariable: 'root', + itemSelector: selectByClass(dataset), + sourceFor: registrySource(new URL('https://registry.example.org/sparql')), + }, + // …the object-grain types keep reading the dataset’s distribution. + ], +}); +``` + +`searchIndexerPipeline` exposes the same routing as `registryTypes`, by type +name: + +```typescript +searchIndexerPipeline({ + schema, + datasets, + writerFor, + registryTypes: { + endpoint: new URL('https://registry.example.org/sparql'), + names: ['Dataset', 'Publisher'], + }, +}); +``` + +Same CONSTRUCT generator, same framing, same projection, same writers – only +the source differs. Two properties make it work: + +- **The scoping is what keeps the stage per-dataset.** A register holds every + registration, so an unscoped stage would index the whole catalogue once per + dataset processed. `registrySource` scopes selection _and_ extraction to the + dataset’s graph, so one pass sees exactly one registration. It presumes the + register names each registration’s graph after the dataset IRI, which is how + a DCAT register that crawls per registration stores it. +- **The hops come from the declared paths.** No CBD rule and no hardcoded + predicate list: a `Publisher` root type with `label` at + `` resolves through the same mechanism as any + other type, because a registration describes its publisher inside its own + graph. + +Routing is deployment topology, deliberately kept out of the `SearchType`: a +type is defined by its `class`, not by where its triples come from, so the same +declaration serves a deployment that sources it differently. + ## Per-stage tuning Each `SearchStageType` entry carries three knobs beyond its selector and diff --git a/packages/pipeline/src/sparql/graph.ts b/packages/pipeline/src/sparql/graph.ts index fbbdea2c..83365ff7 100644 --- a/packages/pipeline/src/sparql/graph.ts +++ b/packages/pipeline/src/sparql/graph.ts @@ -1,16 +1,32 @@ -import { AstFactory, type QueryConstruct } from '@traqula/rules-sparql-1-1'; +import { assertSafeIri } from '@lde/dataset'; +import { + AstFactory, + type QueryConstruct, + type QuerySelect, +} from '@traqula/rules-sparql-1-1'; const F = new AstFactory(); /** - * Set the default graph (FROM clause) on a parsed CONSTRUCT query. + * Set the default graph (FROM clause) on a parsed CONSTRUCT or SELECT query. * * Mutates the query in place, replacing any existing FROM clause. + * + * Both query forms scope the same way, and a stage reading a + * {@link Distribution} with a `namedGraph` must scope **both**: were only the + * reader to honour it, an item selector would pick its roots from the whole + * endpoint and the reader would then find nothing for most of them. + * + * The graph IRI is {@link assertSafeIri}-checked before it becomes an `<…>` + * reference: a `namedGraph` is a plain string, typically carried in from + * third-party registry data, and one containing an angle bracket or whitespace + * would break out of the IRI reference and rewrite the query around it. */ export function withDefaultGraph( - query: QueryConstruct, + query: QueryConstruct | QuerySelect, graphIri: string, ): void { + assertSafeIri(graphIri); query.datasets = F.datasetClauses( [{ clauseType: 'default', value: F.termNamed(F.gen(), graphIri) }], F.gen(), diff --git a/packages/pipeline/src/sparql/selector.ts b/packages/pipeline/src/sparql/selector.ts index 0e6da10a..bd561d50 100644 --- a/packages/pipeline/src/sparql/selector.ts +++ b/packages/pipeline/src/sparql/selector.ts @@ -9,6 +9,7 @@ import { type TermVariable, } from '@traqula/rules-sparql-1-1'; import type { ItemSelector, SelectOptions } from '../stage.js'; +import { withDefaultGraph } from './graph.js'; import type { VariableBindings } from './reader.js'; import { ConstantTimeoutPolicy, @@ -34,7 +35,7 @@ export interface SparqlItemSelectorOptions { * SELECT query projecting at least one named variable. * * A `LIMIT` clause in the query overrides the stage's `batchSize` as the - * page size — use this when the SPARQL endpoint enforces a result limit. + * page size – use this when the SPARQL endpoint enforces a result limit. * It does **not** cap the total number of bindings the selector yields; * pagination continues with `OFFSET` until the source is exhausted. Use * {@link maxResults} to cap the total. @@ -42,7 +43,7 @@ export interface SparqlItemSelectorOptions { query: string; /** * Maximum number of bindings the selector yields across all pages. - * Use this for sampling — “give me at most N items, don’t walk the full + * Use this for sampling – “give me at most N items, don’t walk the full * source”. Independent of {@link query}’s `LIMIT`, which controls page * size. Pagination stops as soon as `maxResults` bindings have been * yielded. Must not be negative. @@ -64,8 +65,12 @@ export interface SparqlItemSelectorOptions { * pagination early. Dropped rows are still fetched; to skip them at the * endpoint, filter in the query itself (e.g. `FILTER(isIRI(?s))`). * - * The endpoint URL comes from the {@link Distribution} passed to {@link select}. - * Pagination is an internal detail — consumers iterate binding rows directly. + * The endpoint URL comes from the {@link Distribution} passed to {@link select}, + * as does the graph to scope to: a distribution declaring a `namedGraph` selects + * within that graph (a `FROM` clause), the same scoping `SparqlConstructReader` + * applies to the extraction. The two must scope alike, or the roots come from a + * wider source than the quads read for them. + * Pagination is an internal detail – consumers iterate binding rows directly. * * The page size (results per SPARQL request) is determined by, in order: * 1. A `LIMIT` clause in the selector query (for endpoints with hard result limits) @@ -129,6 +134,12 @@ export class SparqlItemSelector implements ItemSelector { ); } const endpoint = distribution.accessUrl!; + // Per call, so concurrent selections over different datasets cannot see + // each other's graph scope or page window. + const query = structuredClone(this.parsed); + if (distribution.namedGraph) { + withDefaultGraph(query, distribution.namedGraph); + } const policy = options?.timeout ?? defaultTimeoutPolicy; let offset = 0; let totalFetched = 0; @@ -148,12 +159,12 @@ export class SparqlItemSelector implements ItemSelector { offset === 0 || totalFetched > totalYielded ? basePageSize : Math.min(basePageSize, remaining); - this.parsed.solutionModifiers.limitOffset = F.solutionModifierLimitOffset( + query.solutionModifiers.limitOffset = F.solutionModifierLimitOffset( effectivePageSize, offset, F.gen(), ); - const paginatedQuery = generator.generate(this.parsed); + const paginatedQuery = generator.generate(query); const stream = await this.fetchBindingsWithPolicy( endpoint, diff --git a/packages/pipeline/src/stage.ts b/packages/pipeline/src/stage.ts index e735aada..748b1fe1 100644 --- a/packages/pipeline/src/stage.ts +++ b/packages/pipeline/src/stage.ts @@ -76,9 +76,7 @@ export interface AttachedReader { /** One or more readers, each optionally carrying attached transforms. */ export type StageReaders = - | Reader - | AttachedReader - | (Reader | AttachedReader)[]; + Reader | AttachedReader | (Reader | AttachedReader)[]; /** A reader paired with its attached transforms, normalised to an array. */ interface NormalizedReader { @@ -126,6 +124,35 @@ export interface StageOptions { * @default 10 */ maxConcurrency?: number; + /** + * Where this stage reads, when that is **not** the dataset’s own + * distribution: given the dataset and the distribution the pipeline resolved + * for it, return the {@link Distribution} this stage should use instead. + * + * One hook covers the whole stage – the item selector and every reader are + * handed the same substitute – because a stage that reads elsewhere must + * *select* there too, or its roots and its quads come from different sources. + * + * The dataset is passed so the substitute can be derived per dataset. The + * motivating case is a stage sourced from the dataset registry rather than + * from the dataset’s data: the registry endpoint is fixed, but the graph to + * scope to is the dataset in hand. + * + * ```ts + * sourceFor: (dataset) => { + * const source = new Distribution(registryEndpoint, SPARQL_MEDIA_TYPE); + * source.namedGraph = dataset.iri.toString(); + * return source; + * } + * ``` + * + * Omit it – the default – to read the dataset’s own distribution. + * + * A **chained child** stage may not declare one: a child is handed its + * parent’s output as its distribution, so substituting a source would discard + * the chain it exists to continue. The parent’s constructor rejects that. + */ + sourceFor?: (dataset: Dataset, distribution: Distribution) => Distribution; /** Child stages that chain off this stage's output. */ stages?: Stage[]; /** @@ -175,6 +202,9 @@ export class Stage { readonly stages: readonly Stage[]; /** Whether an empty result is treated as a hard failure. @see {@link StageOptions.expectsOutput} */ readonly expectsOutput: boolean; + /** Whether this stage reads a source of its own rather than the dataset’s + * resolved distribution. @see {@link StageOptions.sourceFor} */ + readonly sourcesOwnData: boolean; private readonly readers: NormalizedReader[]; private readonly itemSelector?: ItemSelector; private readonly batchSize: number; @@ -182,6 +212,7 @@ export class Stage { private readonly validation?: StageOptions['validation']; private readonly project?: BatchTransform; private readonly queueCapacity?: number; + private readonly sourceFor?: StageOptions['sourceFor']; constructor(options: StageOptions) { if (options.project && !options.itemSelector) { @@ -194,6 +225,16 @@ export class Stage { `Stage '${options.name}': 'project' cannot combine with chained 'stages' – a chained stage serializes to N-Triples, which a projected item cannot.`, ); } + // A child is handed its parent's output as its distribution; a `sourceFor` + // would substitute that away, silently reading elsewhere instead of + // continuing the chain. Caught here rather than at run time, where it looks + // like a chain that mysteriously produced nothing. + const sourcedChild = options.stages?.find((child) => child.sourcesOwnData); + if (sourcedChild) { + throw new Error( + `Stage '${options.name}': chained stage '${sourcedChild.name}' declares 'sourceFor' – a chained stage reads its parent's output, so it cannot source its own data.`, + ); + } this.name = options.name; this.stages = options.stages ?? []; this.readers = normalizeReaders(options.readers); @@ -204,6 +245,8 @@ export class Stage { this.expectsOutput = options.expectsOutput ?? false; this.project = options.project; this.queueCapacity = options.queueCapacity; + this.sourceFor = options.sourceFor; + this.sourcesOwnData = options.sourceFor !== undefined; } /** The validator for this stage, if configured. */ @@ -213,11 +256,16 @@ export class Stage { async run( dataset: Dataset, - distribution: Distribution, + resolved: Distribution, writer: DatasetWriter, options?: RunOptions, ): Promise { const timeout = options?.timeout; + // Substituted once, before anything reads: the selector and the readers + // must agree on where this stage's data comes from. + const distribution = this.sourceFor + ? this.sourceFor(dataset, resolved) + : resolved; if (this.itemSelector) { return this.runWithSelector( this.itemSelector.select(distribution, this.batchSize, { diff --git a/packages/pipeline/test/sparql/graph.test.ts b/packages/pipeline/test/sparql/graph.test.ts index 61d7ca9f..90f263c8 100644 --- a/packages/pipeline/test/sparql/graph.test.ts +++ b/packages/pipeline/test/sparql/graph.test.ts @@ -27,6 +27,34 @@ describe('withDefaultGraph', () => { ]); }); + it('scopes a SELECT the same way, for an item selector', () => { + const query = parser.parse( + 'SELECT ?s WHERE { ?s ?p ?o }', + ) as unknown as QueryConstruct; + + withDefaultGraph(query, 'http://example.org/graph'); + + expect(query.datasets.clauses[0]).toMatchObject({ + clauseType: 'default', + value: expect.objectContaining({ value: 'http://example.org/graph' }), + }); + }); + + it('rejects a graph IRI that would break out of the IRI reference', () => { + // A `namedGraph` is a plain string, typically carried in from third-party + // registry data; unchecked, one of these rewrites the query around it. + for (const unsafe of [ + 'http://example.org/a>b', + 'http://example.org/a b', + 'http://example.org/g> WHERE { ?s ?p ?o } #', + ]) { + const query = parseConstruct('CONSTRUCT { ?s ?p ?o } WHERE { ?s ?p ?o }'); + expect(() => withDefaultGraph(query, unsafe)).toThrow( + /unsafe characters/, + ); + } + }); + it('replaces an existing FROM clause', () => { const query = parseConstruct( 'CONSTRUCT { ?s ?p ?o } FROM WHERE { ?s ?p ?o }', diff --git a/packages/pipeline/test/sparql/selector.test.ts b/packages/pipeline/test/sparql/selector.test.ts index 397dbf4f..f1a3a12a 100644 --- a/packages/pipeline/test/sparql/selector.test.ts +++ b/packages/pipeline/test/sparql/selector.test.ts @@ -649,7 +649,7 @@ describe('SparqlItemSelector', () => { fetcher: mockFetcher as never, }); - // No policy supplied at construction or per call — pagination still + // No policy supplied at construction or per call – pagination still // works against the module-level default policy. const rows: VariableBindings[] = []; for await (const row of selector.select(distribution, 10)) { @@ -659,4 +659,98 @@ describe('SparqlItemSelector', () => { expect(mockFetcher.fetchBindings).toHaveBeenCalledTimes(1); }); }); + + describe('graph scoping', () => { + /** A distribution scoped to one named graph, as a registry-sourced stage + * builds per dataset. */ + function scopedTo(graph: string): Distribution { + const scoped = Distribution.sparql(new URL('http://example.com/sparql')); + scoped.namedGraph = graph; + return scoped; + } + + it('selects within the distribution’s named graph', async () => { + const { fetcher, queries } = pagedFetcher([ + [{ uri: namedNode('http://example.com/1') }], + ]); + const selector = new SparqlItemSelector({ + query, + fetcher: fetcher as never, + }); + + const rows: VariableBindings[] = []; + for await (const row of selector.select( + scopedTo('http://example.com/graph'), + 10, + )) { + rows.push(row); + } + + expect(rows).toHaveLength(1); + expect(queries[0]).toContain('FROM '); + }); + + it('leaves the query unscoped when the distribution names no graph', async () => { + const { fetcher, queries } = pagedFetcher([[]]); + const selector = new SparqlItemSelector({ + query, + fetcher: fetcher as never, + }); + + await selectAll(selector, 10); + + expect(queries[0]).not.toContain('FROM'); + }); + + it('keeps the scope on every page, not just the first', async () => { + // Pagination rewrites the query per page; were the scope applied to the + // page rather than the selection, page 2 onwards would silently widen to + // the whole endpoint. + const { fetcher, queries } = pagedFetcher([ + [ + { uri: namedNode('http://example.com/1') }, + { uri: namedNode('http://example.com/2') }, + ], + [{ uri: namedNode('http://example.com/3') }], + ]); + const selector = new SparqlItemSelector({ + query, + fetcher: fetcher as never, + }); + + for await (const _row of selector.select( + scopedTo('http://example.com/graph'), + 2, + )) { + // consume + } + + expect(queries).toHaveLength(2); + expect(queries[1]).toContain('FROM '); + expect(queries[1]).toMatch(/OFFSET\s+2/); + }); + + it('scopes each selection independently', async () => { + // One selector instance serves every dataset, so a per-call scope must + // not leak into the next call – nor stay behind once a call is unscoped. + const { fetcher, queries } = pagedFetcher([[], [], []]); + const selector = new SparqlItemSelector({ + query, + fetcher: fetcher as never, + }); + + await selectAll(selector, 10); + for await (const _row of selector.select(scopedTo('urn:graph:1'), 10)) { + // consume + } + for await (const _row of selector.select(scopedTo('urn:graph:2'), 10)) { + // consume + } + + expect(queries[0]).not.toContain('FROM'); + expect(queries[1]).toContain('FROM '); + expect(queries[2]).toContain('FROM '); + expect(queries[2]).not.toContain('urn:graph:1'); + }); + }); }); diff --git a/packages/pipeline/test/stage.test.ts b/packages/pipeline/test/stage.test.ts index 2bb0880c..e80f5ac1 100644 --- a/packages/pipeline/test/stage.test.ts +++ b/packages/pipeline/test/stage.test.ts @@ -650,19 +650,15 @@ describe('Stage', () => { ): Validator { const violations = options.violations ?? (options.conforms ? 0 : 1); return { - validate: vi.fn( - async (): Promise => ({ - conforms: options.conforms, - violations, - }), - ), - report: vi.fn( - async (): Promise => ({ - conforms: options.conforms, - violations, - quadsValidated: 0, - }), - ), + validate: vi.fn(async (): Promise => ({ + conforms: options.conforms, + violations, + })), + report: vi.fn(async (): Promise => ({ + conforms: options.conforms, + violations, + quadsValidated: 0, + })), }; } @@ -794,13 +790,11 @@ describe('Stage', () => { writeOrder.push('validate'); return { conforms: true, violations: 0 }; }), - report: vi.fn( - async (): Promise => ({ - conforms: true, - violations: 0, - quadsValidated: 2, - }), - ), + report: vi.fn(async (): Promise => ({ + conforms: true, + violations: 0, + quadsValidated: 2, + })), }; const writer: DatasetWriter = { @@ -834,13 +828,11 @@ describe('Stage', () => { writeOrder.push('validate'); return { conforms: false, violations: 1 }; }), - report: vi.fn( - async (): Promise => ({ - conforms: false, - violations: 1, - quadsValidated: 1, - }), - ), + report: vi.fn(async (): Promise => ({ + conforms: false, + violations: 1, + quadsValidated: 1, + })), }; const writer: DatasetWriter = { @@ -1224,4 +1216,126 @@ describe('Stage', () => { ).toThrow(/cannot combine with chained 'stages'/); }); }); + + describe('sourceFor', () => { + /** The registry-sourced shape: a fixed endpoint, scoped to the graph the + * dataset in hand names. */ + function registrySource(dataset: Dataset): Distribution { + const source = Distribution.sparql( + new URL('http://registry.example.org/sparql'), + ); + source.namedGraph = dataset.iri.toString(); + return source; + } + + it('hands readers the substituted distribution', async () => { + const reader = capturingExecutor([q1]); + const stage = new Stage({ + name: 'registry', + readers: reader, + sourceFor: registrySource, + }); + + await stage.run(dataset, distribution, collectingWriter()); + + const source = reader.read.mock.calls[0][1] as Distribution; + expect(source.accessUrl.toString()).toBe( + 'http://registry.example.org/sparql', + ); + expect(source.namedGraph).toBe('http://example.org/dataset'); + }); + + it('hands the item selector the same substituted distribution', async () => { + // Selection and extraction must agree on the source: roots selected from + // one endpoint and quads read from another join on nothing. + const seen: Distribution[] = []; + const selector: ItemSelector = { + async *select(distribution) { + seen.push(distribution); + yield { s: namedNode('http://example.org/s1') }; + }, + }; + const reader = capturingExecutor([q1]); + const stage = new Stage({ + name: 'registry', + readers: reader, + itemSelector: selector, + sourceFor: registrySource, + }); + + await stage.run(dataset, distribution, collectingWriter()); + + expect(seen[0]).toBe(reader.read.mock.calls[0][1]); + expect(seen[0].accessUrl.toString()).toBe( + 'http://registry.example.org/sparql', + ); + }); + + it('receives the resolved distribution, so it can defer to it', async () => { + const reader = capturingExecutor([q1]); + const stage = new Stage({ + name: 'passthrough', + readers: reader, + sourceFor: (_dataset, resolved) => resolved, + }); + + await stage.run(dataset, distribution, collectingWriter()); + + expect(reader.read.mock.calls[0][1]).toBe(distribution); + }); + + it('rejects a chained child that sources its own data', () => { + // A child reads its parent’s output as its distribution, so a sourceFor + // would substitute the chain away – a chain that mysteriously produces + // nothing, if it were only caught at run time. + const child = new Stage({ + name: 'child', + readers: mockExecutor([]), + sourceFor: registrySource, + }); + expect( + () => + new Stage({ + name: 'parent', + readers: mockExecutor([q1]), + stages: [child], + }), + ).toThrow(/chained stage 'child' declares 'sourceFor'/); + }); + + it('allows a chaining parent to source its own data', () => { + const child = new Stage({ name: 'child', readers: mockExecutor([]) }); + expect( + () => + new Stage({ + name: 'parent', + readers: mockExecutor([q1]), + sourceFor: registrySource, + stages: [child], + }), + ).not.toThrow(); + }); + + it('reports whether a stage sources its own data', () => { + expect( + new Stage({ name: 'plain', readers: mockExecutor([]) }).sourcesOwnData, + ).toBe(false); + expect( + new Stage({ + name: 'registry', + readers: mockExecutor([]), + sourceFor: registrySource, + }).sourcesOwnData, + ).toBe(true); + }); + + it('reads the dataset’s own distribution when omitted', async () => { + const reader = capturingExecutor([q1]); + const stage = new Stage({ name: 'plain', readers: reader }); + + await stage.run(dataset, distribution, collectingWriter()); + + expect(reader.read.mock.calls[0][1]).toBe(distribution); + }); + }); }); diff --git a/packages/pipeline/vite.config.ts b/packages/pipeline/vite.config.ts index b43f8775..c26ff1a2 100644 --- a/packages/pipeline/vite.config.ts +++ b/packages/pipeline/vite.config.ts @@ -11,10 +11,10 @@ export default mergeConfig( coverage: { thresholds: { autoUpdate: true, - functions: 97.45, - lines: 97.12, - branches: 91.9, - statements: 96.67, + functions: 97.46, + lines: 97.15, + branches: 91.98, + statements: 96.7, }, }, }, diff --git a/packages/search-indexer/src/config.ts b/packages/search-indexer/src/config.ts index 5a6a58d5..14936a4e 100644 --- a/packages/search-indexer/src/config.ts +++ b/packages/search-indexer/src/config.ts @@ -12,6 +12,15 @@ export interface IndexerConfig { /** SPARQL endpoint of the DCAT dataset registry the selection queries * (`REGISTRY_ENDPOINT`). */ readonly registryEndpoint: URL; + /** + * Root types extracted from {@link registryEndpoint} instead of from each + * dataset’s own distribution (`REGISTRY_ROOT_TYPES`, whitespace- or + * comma-separated {@link SearchType.name}s) – the dataset’s own description + * and what hangs off it, which live in the register rather than in the data a + * publisher ships. Empty when unset: every type reads the distribution, the + * behaviour before this setting existed. + */ + readonly registryRootTypes: readonly string[]; /** Dataset selection within the registry: explicit IRIs (`DATASETS`), * search criteria (`DATASET_CRITERIA`), or every dataset when neither is * set. */ @@ -145,6 +154,7 @@ export function configFromEnvironment( schemaModulePath: environment['SCHEMA_MODULE'] ?? '/config/search-schema.mjs', registryEndpoint, + registryRootTypes: names(environment['REGISTRY_ROOT_TYPES']), datasetCriteria, typesense: { host: required('TYPESENSE_HOST'), @@ -165,6 +175,17 @@ export function configFromEnvironment( return config; } +/** + * A whitespace- or comma-separated list variable, as a list. The names + * themselves are validated against the mounted schema when the pipeline is + * wired – config has no schema to check them against. + */ +function names(value: string | undefined): readonly string[] { + return value === undefined + ? [] + : value.split(/[\s,]+/).filter((name) => name.length > 0); +} + /** `DATASETS` (whitespace-separated IRIs) or `DATASET_CRITERIA` (a JSON * object) – never both; neither selects the whole registry. */ function criteriaFromEnvironment( diff --git a/packages/search-indexer/src/indexer.ts b/packages/search-indexer/src/indexer.ts index 88d27aa6..c3f01aca 100644 --- a/packages/search-indexer/src/indexer.ts +++ b/packages/search-indexer/src/indexer.ts @@ -40,6 +40,13 @@ export async function createSearchIndexer( datasets: datasetSelectorFrom(config), distributionResolver: distributionResolverFrom(config), writerFor: writerFactoryFrom(client, config), + registryTypes: + config.registryRootTypes.length > 0 + ? { + endpoint: config.registryEndpoint, + names: config.registryRootTypes, + } + : undefined, provenanceStore: config.provenance ? new FileProvenanceStore({ path: config.provenance.path }) : undefined, diff --git a/packages/search-indexer/test/config.test.ts b/packages/search-indexer/test/config.test.ts index 0fe61174..5d9111d0 100644 --- a/packages/search-indexer/test/config.test.ts +++ b/packages/search-indexer/test/config.test.ts @@ -13,6 +13,7 @@ describe('configFromEnvironment', () => { expect(config).toEqual({ schemaModulePath: '/config/search-schema.mjs', registryEndpoint: new URL('https://registry.example.org/sparql'), + registryRootTypes: [], datasetCriteria: {}, typesense: { host: 'typesense.internal', @@ -35,6 +36,7 @@ describe('configFromEnvironment', () => { TYPESENSE_PROTOCOL: 'https', REBUILD_MODE: 'blue-green', COLLECTION_PREFIX: 'staging_', + REGISTRY_ROOT_TYPES: 'Dataset, Publisher', QLEVER_IMAGE: 'adfreiburg/qlever:latest', IMPORT_STRATEGY: 'import', DATA_DIR: '/mnt/data', @@ -42,6 +44,7 @@ describe('configFromEnvironment', () => { expect(config).toEqual({ schemaModulePath: '/mnt/schema.mjs', registryEndpoint: new URL('https://registry.example.org/sparql'), + registryRootTypes: ['Dataset', 'Publisher'], datasetCriteria: {}, typesense: { host: 'typesense.internal', diff --git a/packages/search-indexer/test/indexer.test.ts b/packages/search-indexer/test/indexer.test.ts index 922725a6..b2487551 100644 --- a/packages/search-indexer/test/indexer.test.ts +++ b/packages/search-indexer/test/indexer.test.ts @@ -34,6 +34,31 @@ describe('createSearchIndexer', () => { expect(pipeline).toBeInstanceOf(Pipeline); }); + it('routes the configured root types to the registry endpoint', async () => { + const pipeline = await createSearchIndexer( + configFromEnvironment({ + ...minimal, + SCHEMA_MODULE: fixture, + REGISTRY_ROOT_TYPES: 'Dataset', + }), + ); + expect(pipeline).toBeInstanceOf(Pipeline); + }); + + it('fails the boot on a registry root type the schema does not declare', async () => { + // At boot, not per dataset: the type would otherwise read the dataset’s + // distribution and ship an empty collection. + await expect( + createSearchIndexer( + configFromEnvironment({ + ...minimal, + SCHEMA_MODULE: fixture, + REGISTRY_ROOT_TYPES: 'Publisher', + }), + ), + ).rejects.toThrowError(/Unknown registry root type\(s\) “Publisher”/); + }); + it('fails the boot on an unloadable schema module, naming the path', async () => { await expect( createSearchIndexer( diff --git a/packages/search-pipeline/src/index.ts b/packages/search-pipeline/src/index.ts index a6d46476..a3e67e9b 100644 --- a/packages/search-pipeline/src/index.ts +++ b/packages/search-pipeline/src/index.ts @@ -1,6 +1,10 @@ export { searchIndexerPipeline } from './search-indexer-pipeline.js'; export type { SearchIndexerPipelineOptions } from './search-indexer-pipeline.js'; -export { searchStages, selectByClass } from './search-stages.js'; +export { + registrySource, + searchStages, + selectByClass, +} from './search-stages.js'; export type { SearchStagesOptions, SearchStageType } from './search-stages.js'; export { extractionQuery, extractionQueryString } from './extraction.js'; export type { ExtractionOptions } from './extraction.js'; diff --git a/packages/search-pipeline/src/search-indexer-pipeline.ts b/packages/search-pipeline/src/search-indexer-pipeline.ts index 0a1ceefd..7101ecf5 100644 --- a/packages/search-pipeline/src/search-indexer-pipeline.ts +++ b/packages/search-pipeline/src/search-indexer-pipeline.ts @@ -10,7 +10,11 @@ import { } from '@lde/pipeline'; import type { RootType, SearchDocument, SearchSchema } from '@lde/search'; import { searchIndexWriter } from './search-index-writer.js'; -import { searchStages, selectByClass } from './search-stages.js'; +import { + registrySource, + searchStages, + selectByClass, +} from './search-stages.js'; import type { TypedSearchDocument } from './typed-search-document.js'; /** Options for {@link searchIndexerPipeline}. */ @@ -49,6 +53,25 @@ export interface SearchIndexerPipelineOptions { searchType: RootType, schema: SearchSchema, ) => Writer; + /** + * Root types whose triples come from the **dataset registry** rather than + * from each dataset’s own distribution – typically the dataset description + * itself and what hangs off it, which live in the register because + * registering a dataset is submitting a description + * ({@link registrySource}). + * + * Routing is deployment topology, so it is configured here and not declared + * on the {@link SearchType}: a type is defined by its `class`, not by where + * its triples come from. Name the types by their `name`; an unknown name + * throws here, at wiring time, rather than yielding an empty collection at + * run time. + */ + registryTypes?: { + /** The registry’s SPARQL endpoint – where these types read. */ + endpoint: URL; + /** {@link SearchType.name}s of the root types to source from it. */ + names: readonly string[]; + }; /** * Optional per-dataset processing memory: skip a dataset whose source * fingerprint and {@link pipelineVersion} both match the stored record. @@ -104,7 +127,12 @@ export interface SearchIndexerPipelineOptions { export function searchIndexerPipeline( options: SearchIndexerPipelineOptions, ): Pipeline { - const { schema, datasets } = options; + const { schema, datasets, registryTypes } = options; + const fromRegistry = registryTypeNames(schema, registryTypes?.names); + const sourceFor = + registryTypes === undefined + ? undefined + : registrySource(registryTypes.endpoint); return new Pipeline({ datasetSelector: Array.isArray(datasets) ? new ManualDatasetSelection(datasets) @@ -116,6 +144,7 @@ export function searchIndexerPipeline( searchType, rootVariable: 'root', itemSelector: selectByClass(searchType), + sourceFor: fromRegistry.has(searchType.name) ? sourceFor : undefined, })), }), writers: searchIndexWriter({ schema, writerFor: options.writerFor }), @@ -124,3 +153,30 @@ export function searchIndexerPipeline( reporter: options.reporter, }); } + +/** + * The declared registry-sourced type names, validated against the schema. An + * unrecognised name is a configuration mistake that would otherwise be silent – + * the type would read the dataset’s distribution, find no dataset description + * there, and ship an empty collection – so it throws at wiring time. + */ +function registryTypeNames( + schema: SearchSchema, + declaredNames: readonly string[] | undefined, +): ReadonlySet { + const names = new Set(declaredNames ?? []); + const declared = new Set( + [...schema.values()].map((searchType) => searchType.name), + ); + const unknown = [...names].filter((name) => !declared.has(name)); + if (unknown.length > 0) { + throw new Error( + `Unknown registry root type(s) ${unknown + .map((name) => `“${name}”`) + .join(', ')}; the schema declares ${[...declared] + .map((name) => `“${name}”`) + .join(', ')}.`, + ); + } + return names; +} diff --git a/packages/search-pipeline/src/search-stages.ts b/packages/search-pipeline/src/search-stages.ts index 055ca176..b67b2ef8 100644 --- a/packages/search-pipeline/src/search-stages.ts +++ b/packages/search-pipeline/src/search-stages.ts @@ -1,8 +1,10 @@ +import { Distribution } from '@lde/dataset'; import { SparqlConstructReader, SparqlItemSelector, Stage, type ItemSelector, + type StageOptions, type StageReaders, } from '@lde/pipeline'; import { projectRoots, type RootType, type SearchSchema } from '@lde/search'; @@ -42,6 +44,13 @@ export interface SearchStageType { * readers, or attach transforms. */ readers?: StageReaders; + /** + * Where this type’s stage reads, when that is not the dataset’s own + * distribution – see {@link StageOptions.sourceFor}. {@link registrySource} + * builds the one that matters here: a root type described by the **dataset + * registry** rather than by the dataset’s data. + */ + sourceFor?: StageOptions['sourceFor']; /** * Roots (and so documents) per batch – the memory bound. Under a root-bound * selector it moves memory and request count, never output. @@ -112,6 +121,7 @@ export function searchStages( name: searchType.name, readers, itemSelector: type.itemSelector, + sourceFor: type.sourceFor, batchSize: type.batchSize, maxConcurrency: type.maxConcurrency, queueCapacity: type.queueCapacity, @@ -170,3 +180,48 @@ export function selectByClass( query: `SELECT ?${rootVariable} WHERE { ?${rootVariable} a <${searchType.class}> FILTER(!isBlank(?${rootVariable})) }`, }); } + +/** + * A {@link SearchStageType.sourceFor} that reads a root type from the **dataset + * registry** instead of from the dataset’s own data – the endpoint fixed, + * scoped to the graph the dataset in hand names. + * + * A dataset’s description is governed by a different application profile from + * the objects it contains, and it lives in the register, not in the + * distribution: registering a dataset *is* submitting a description, so the + * register is the one source that covers every dataset a pipeline can select. + * Nothing obliges a publisher to ship a description of the dataset in its own + * dump. + * + * The graph scoping is what keeps the stage per-dataset. A register holds every + * registration, so an unscoped stage would index the whole catalogue for each + * dataset processed; scoped, each pass sees exactly the one registration – and + * the roots a `selectByClass` finds inside that graph are that dataset’s own. + * It presumes the register names each registration’s graph after the dataset + * IRI, which is how a DCAT register that crawls per registration stores it (the + * NDE Dataset Register does so for every registration it holds). + * + * Routing is deployment topology, deliberately kept out of the + * {@link SearchType}: a type is defined by its `class`, not by where its + * triples come from, so the same declaration serves a deployment that sources + * it differently. + * + * ```ts + * searchStages({ + * schema, + * types: [...schema.values()].map((searchType) => ({ + * searchType, + * rootVariable: 'root', + * itemSelector: selectByClass(searchType), + * sourceFor: registryTypeNames.has(searchType.name) + * ? registrySource(registryEndpoint) + * : undefined, + * })), + * }); + * ``` + */ +export function registrySource( + endpoint: URL, +): NonNullable { + return (dataset) => Distribution.sparql(endpoint, dataset.iri.toString()); +} diff --git a/packages/search-pipeline/test/search-indexer-pipeline.test.ts b/packages/search-pipeline/test/search-indexer-pipeline.test.ts index 6ca058a6..54526589 100644 --- a/packages/search-pipeline/test/search-indexer-pipeline.test.ts +++ b/packages/search-pipeline/test/search-indexer-pipeline.test.ts @@ -1,5 +1,6 @@ import { describe, expect, it, vi } from 'vitest'; -import { Dataset } from '@lde/dataset'; +import nock from 'nock'; +import { Dataset, Distribution } from '@lde/dataset'; import { ManualDatasetSelection, Pipeline, @@ -93,4 +94,96 @@ describe('searchIndexerPipeline', () => { }), ).toThrow(/pipelineVersion is required/); }); + + describe('registryTypes', () => { + const registry = { + endpoint: new URL('http://registry.example.org/sparql'), + names: ['Dataset'], + }; + + it('routes only the named types to the registry endpoint', async () => { + // Observable end to end: the routed type selects and extracts against the + // registry, scoped to the dataset’s graph, while the unrouted one stays on + // the dataset’s own distribution. + const endpoints: string[] = []; + const queries: string[] = []; + for (const host of [ + 'http://registry.example.org', + 'http://data.example.org', + ]) { + nock(host) + .post('/sparql') + .times(4) + .reply( + 200, + (_uri, body) => { + endpoints.push(host); + queries.push( + decodeURIComponent(String(body).replace(/\+/g, ' ')), + ); + return { head: { vars: ['root'] }, results: { bindings: [] } }; + }, + { 'Content-Type': 'application/sparql-results+json' }, + ); + } + + const dataset = new Dataset({ + iri: new URL('http://example.org/dataset/1'), + distributions: [ + Distribution.sparql(new URL('http://data.example.org/sparql')), + ], + }); + await searchIndexerPipeline({ + schema, + datasets: [dataset], + writerFor: () => engineWriter, + registryTypes: registry, + }).run(); + + const registryQueries = queries.filter( + (_query, index) => endpoints[index] === 'http://registry.example.org', + ); + expect(registryQueries).toHaveLength(1); + expect(registryQueries[0]).toContain(DATASET); + // Scoped to the registration’s own graph, so a per-dataset pass never + // walks the whole register. + expect(registryQueries[0]).toContain( + 'FROM ', + ); + + const dataQueries = queries.filter( + (_query, index) => endpoints[index] === 'http://data.example.org', + ); + expect(dataQueries.some((query) => query.includes(ORGANIZATION))).toBe( + true, + ); + // The dataset’s own endpoint is never asked for the registry-sourced + // type, and nothing sent there is graph-scoped. + expect(dataQueries.some((query) => query.includes(DATASET))).toBe(false); + expect(dataQueries.some((query) => query.includes('FROM'))).toBe(false); + }); + + it('leaves every type on the distribution when unset', () => { + expect(() => + searchIndexerPipeline({ + schema, + datasets: [], + writerFor: () => engineWriter, + }), + ).not.toThrow(); + }); + + it('rejects a name the schema does not declare', () => { + // Otherwise silent: the type would read the dataset’s distribution, find + // no dataset description there and ship an empty collection. + expect(() => + searchIndexerPipeline({ + schema, + datasets: [], + writerFor: () => engineWriter, + registryTypes: { ...registry, names: ['Datasets'] }, + }), + ).toThrow(/Unknown registry root type\(s\) “Datasets”/); + }); + }); }); diff --git a/packages/search-pipeline/test/search-stages.test.ts b/packages/search-pipeline/test/search-stages.test.ts index 6a14318b..42f7f102 100644 --- a/packages/search-pipeline/test/search-stages.test.ts +++ b/packages/search-pipeline/test/search-stages.test.ts @@ -13,7 +13,11 @@ import { } from '@lde/pipeline'; import { projectRoots, searchSchema, type RootType } from '@lde/search'; import { irAlias } from '@lde/search/adapter'; -import { searchStages, selectByClass } from '../src/search-stages.js'; +import { + registrySource, + searchStages, + selectByClass, +} from '../src/search-stages.js'; import type { TypedSearchDocument } from '../src/typed-search-document.js'; const { namedNode, literal, quad } = DataFactory; @@ -234,6 +238,123 @@ describe('selectByClass', () => { }); }); +describe('registrySource', () => { + const REGISTRY = new URL('http://registry.example.org/sparql'); + + it('reads the registry endpoint, scoped to the dataset’s own graph', () => { + const source = registrySource(REGISTRY)(dataset, distribution); + + expect(source.accessUrl.toString()).toBe(REGISTRY.toString()); + // The register names each registration’s graph after the dataset IRI, so + // scoping to it is what keeps a per-dataset pass from indexing the whole + // catalogue. + expect(source.namedGraph).toBe('http://example.org/dataset/1'); + expect(source.isSparql()).toBe(true); + }); + + it('scopes per dataset, not per pipeline', () => { + const sourceFor = registrySource(REGISTRY); + const other = new Dataset({ + iri: new URL('http://example.org/dataset/2'), + distributions: [], + }); + + expect(sourceFor(dataset, distribution).namedGraph).toBe( + 'http://example.org/dataset/1', + ); + expect(sourceFor(other, distribution).namedGraph).toBe( + 'http://example.org/dataset/2', + ); + }); + + it('routes a type’s selector and readers alike', async () => { + // Selection and extraction must meet the same source: roots selected from + // the registry and quads read from the dataset’s distribution would join on + // nothing. + const seen: Distribution[] = []; + const selector: ItemSelector = { + select: async function* ( + source: Distribution, + ): AsyncIterable { + seen.push(source); + yield { root: namedNode('https://ex/p/1') }; + }, + }; + const reader: Reader = { + read: (_dataset, source, options) => { + seen.push(source); + const quads = (options?.bindings ?? []).map((binding) => + quad( + namedNode(binding.root.value), + namedNode(NAME_ALIAS), + literal('Name'), + ), + ); + return Promise.resolve(stream(quads)); + }, + }; + const [stage] = searchStages({ + schema, + types: [ + { + searchType: person, + rootVariable: 'root', + itemSelector: selector, + readers: reader, + sourceFor: registrySource(REGISTRY), + }, + ], + }); + + const received: TypedSearchDocument[] = []; + await stage.run(dataset, distribution, { + write: async (_dataset, items) => { + for await (const item of items) { + received.push(item); + } + }, + }); + + expect(seen).toHaveLength(2); + expect(seen[0]).toBe(seen[1]); + expect(seen[0].accessUrl.toString()).toBe(REGISTRY.toString()); + expect(received.map((item) => item.document)).toEqual([ + { id: 'https://ex/p/1', name: 'Name' }, + ]); + }); + + it('leaves an unrouted type on the dataset’s own distribution', async () => { + const seen: Distribution[] = []; + const reader: Reader = { + read: (_dataset, source) => { + seen.push(source); + return Promise.resolve(stream([])); + }, + }; + const [stage] = searchStages({ + schema, + types: [ + { + searchType: person, + rootVariable: 'root', + itemSelector: rootsSelector(['https://ex/p/1']), + readers: reader, + }, + ], + }); + + await stage.run(dataset, distribution, { + write: async (_dataset, items) => { + for await (const item of items) { + void item; + } + }, + }); + + expect(seen[0]).toBe(distribution); + }); +}); + describe('memory is bounded by batchSize, not the input (counting, not measuring)', () => { const BATCH_SIZE = 10;