From 6feedfed45db42d181ebab3c9cb3f2be46b708f6 Mon Sep 17 00:00:00 2001 From: David de Boer Date: Thu, 23 Jul 2026 16:06:50 +0200 Subject: [PATCH] feat(search-pipeline): add searchIndexerPipeline convenience - Wire the common object-grain indexer into a ready-to-run Pipeline: one stage per root type in the schema, roots selected by class, extraction generated from the schema, one engine writer per type, with optional provenance and reporting passed through. - Exclude blank-node subjects in selectByClass via FILTER(!isBlank(...)): a blank node has no stable document key, so it can never become a search document and would crash framing. - Document the role-to-package mapping in the search-pipeline README and point the stub READMEs of sparql-server and sparql-importer at sparql-qlever as the concrete adapter. --- AGENTS.md | 2 +- packages/search-pipeline/README.md | 56 +++++++ packages/search-pipeline/src/index.ts | 2 + .../src/search-indexer-pipeline.ts | 119 +++++++++++++++ ...earch-indexer-pipeline.integration.test.ts | 137 ++++++++++++++++++ .../test/search-indexer-pipeline.test.ts | 96 ++++++++++++ packages/sparql-importer/README.md | 11 ++ packages/sparql-server/README.md | 10 ++ 8 files changed, 432 insertions(+), 1 deletion(-) create mode 100644 packages/search-pipeline/src/search-indexer-pipeline.ts create mode 100644 packages/search-pipeline/test/search-indexer-pipeline.integration.test.ts create mode 100644 packages/search-pipeline/test/search-indexer-pipeline.test.ts diff --git a/AGENTS.md b/AGENTS.md index 0024a2b1..2633ba19 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -82,7 +82,7 @@ Each package uses conditional exports with a `development` condition for local d - Test files use `.test.ts` suffix in `test/` directory - Fixtures in `test/fixtures/` - HTTP mocking with Nock -- Tests that start a local SPARQL endpoint (`@lde/local-sparql-endpoint`) must use unique ports across packages to avoid conflicts when Nx runs tests in parallel. Current port allocations: `dataset-registry-client` (3002), `pipeline` sparqlQuery (3001), `pipeline` executor (3003), `pipeline` provenance store (3004), `pipeline-void` namespace-normalization (3005–3006), `search-pipeline` extraction round-trip (3007) +- Tests that start a local SPARQL endpoint (`@lde/local-sparql-endpoint`) must use unique ports across packages to avoid conflicts when Nx runs tests in parallel. Current port allocations: `dataset-registry-client` (3002), `pipeline` sparqlQuery (3001), `pipeline` executor (3003), `pipeline` provenance store (3004), `pipeline-void` namespace-normalization (3005–3006), `search-pipeline` extraction round-trip (3007), `search-pipeline` searchIndexerPipeline end-to-end (3008) ### Key Dependencies diff --git a/packages/search-pipeline/README.md b/packages/search-pipeline/README.md index 222573f1..97a0a16f 100644 --- a/packages/search-pipeline/README.md +++ b/packages/search-pipeline/README.md @@ -34,6 +34,62 @@ The division of labour ([ADR 6](../../docs/decisions/0006-make-the-writer-transa So a search pipeline is **one terminal, N stages**: `new Pipeline({ datasetSelector, stages: searchStages(...), writers: searchIndexWriter(...) })`. +## Quick start: `searchIndexerPipeline` + +For the common **object grain** – index every IRI-identified instance of each +root type’s source class – `searchIndexerPipeline` wires that whole composition +and returns the ready-to-run `Pipeline`. It generates one stage per root type +in the schema, each selecting the type’s non-blank roots by class +(`selectByClass`; a blank node has no stable document key, so it can never +become a document), extracting with the schema-generated CONSTRUCT, and routing +every document to the engine writer for its type’s collection. The consumer +supplies only the domain (the schema, which datasets) and the deployment shell +(the engine writer, the SPARQL/import adapter): + +```typescript +import { searchIndexerPipeline } from '@lde/search-pipeline'; +import { + FileProvenanceStore, + ImportResolver, + SparqlDistributionResolver, +} from '@lde/pipeline'; +import { ConsoleReporter } from '@lde/pipeline-console-reporter'; +import { createQlever } from '@lde/sparql-qlever'; +import { InPlaceRebuild } from '@lde/search-typesense'; + +const pipeline = searchIndexerPipeline({ + schema, + datasets, + // Import each dataset’s data dump into a local QLever and query that. + distributionResolver: new ImportResolver(new SparqlDistributionResolver(), { + ...createQlever({ mode: 'docker', image: 'adfreiburg/qlever:latest' }), + strategy: 'import', + }), + writerFor: (searchType) => new InPlaceRebuild(typesenseClient, searchType), + // Optional: skip datasets whose source and pipeline version are unchanged. + provenanceStore: new FileProvenanceStore({ path: 'data/provenance.json' }), + pipelineVersion: '1', + reporter: new ConsoleReporter(), +}); +await pipeline.run(); +``` + +Each role in that wiring maps to a package: + +| Role | Package | +| -------------------------------------- | ---------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------- | +| Dataset selection | [`@lde/pipeline`](../pipeline) – `Dataset[]` directly, or a `DatasetSelector` such as `RegistrySelector` | +| Distribution resolution and import | [`@lde/pipeline`](../pipeline) `ImportResolver` + [`@lde/sparql-qlever`](../sparql-qlever) `createQlever` (the concrete [`@lde/sparql-server`](../sparql-server)/[`@lde/sparql-importer`](../sparql-importer) adapter) | +| Root selection, extraction, projection | this package (`searchStages` over [`@lde/search`](../search)) | +| Engine writer, one collection per type | [`@lde/search-typesense`](../search-typesense) – `InPlaceRebuild` or `BlueGreenRebuild` | +| Provenance (skip-unchanged) | [`@lde/pipeline`](../pipeline) – `FileProvenanceStore`, or a triplestore-backed store | +| Reporting | [`@lde/pipeline-console-reporter`](../pipeline-console-reporter) `ConsoleReporter` | + +A deployment that needs more – a bespoke root selector (where the entry point +is a domain fact, not a class), per-stage tuning, non-SPARQL readers, or +quad-level plugins – composes the parts directly, as below; the convenience +owns no capability of its own. + ## Usage ```typescript diff --git a/packages/search-pipeline/src/index.ts b/packages/search-pipeline/src/index.ts index 7cb263a0..a6d46476 100644 --- a/packages/search-pipeline/src/index.ts +++ b/packages/search-pipeline/src/index.ts @@ -1,3 +1,5 @@ +export { searchIndexerPipeline } from './search-indexer-pipeline.js'; +export type { SearchIndexerPipelineOptions } from './search-indexer-pipeline.js'; export { searchStages, selectByClass } from './search-stages.js'; export type { SearchStagesOptions, SearchStageType } from './search-stages.js'; export { extractionQuery, extractionQueryString } from './extraction.js'; diff --git a/packages/search-pipeline/src/search-indexer-pipeline.ts b/packages/search-pipeline/src/search-indexer-pipeline.ts new file mode 100644 index 00000000..ac2954c5 --- /dev/null +++ b/packages/search-pipeline/src/search-indexer-pipeline.ts @@ -0,0 +1,119 @@ +import type { Dataset } from '@lde/dataset'; +import { + ManualDatasetSelection, + Pipeline, + type DatasetSelector, + type DistributionResolver, + type ProgressReporter, + type ProvenanceStore, + type Writer, +} 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 type { TypedSearchDocument } from './typed-search-document.js'; + +/** Options for {@link searchIndexerPipeline}. */ +export interface SearchIndexerPipelineOptions { + /** + * The declarative schema driving the indexer: one stage and one engine + * collection per {@link RootType} in it. Reference Types are absent from + * `schema.values()`, so none ever earns a stage or a collection. + */ + schema: SearchSchema; + /** + * Which datasets to index – the deployment’s domain fact. Pass the + * {@link Dataset}s directly, or a {@link DatasetSelector} (e.g. a + * `RegistrySelector`) when selection is dynamic. + */ + datasets: Dataset[] | DatasetSelector; + /** + * How each dataset’s data becomes queryable – the deployment’s engine + * choice. Typically an `ImportResolver` wrapping a + * `SparqlDistributionResolver` with a `@lde/sparql-qlever` `createQlever` + * import path, so data-dump distributions are imported into a local + * endpoint. Defaults to the {@link Pipeline} default (a bare + * `SparqlDistributionResolver`), which serves only datasets that publish a + * live SPARQL endpoint. + */ + distributionResolver?: DistributionResolver; + /** + * The engine writer that owns a given root type’s collection – e.g. a + * `@lde/search-typesense` `InPlaceRebuild` or `BlueGreenRebuild` bound to + * that type. Called once per {@link RootType} in the schema. + */ + writerFor: (searchType: RootType) => Writer; + /** + * Optional per-dataset processing memory: skip a dataset whose source + * fingerprint and {@link pipelineVersion} both match the stored record. + * Requires {@link pipelineVersion}. + */ + provenanceStore?: ProvenanceStore; + /** + * Opaque, consumer-declared version of the indexer’s output-affecting logic + * (schema, projection). Required when {@link provenanceStore} is set – a + * skip-enabled pipeline with no version would silently freeze. + */ + pipelineVersion?: string; + /** + * Observer(s) of pipeline lifecycle events, e.g. a + * `@lde/pipeline-console-reporter` `ConsoleReporter`. + */ + reporter?: ProgressReporter | readonly ProgressReporter[]; +} + +/** + * Wire the common **object-grain** search indexer and return the ready-to-run + * {@link Pipeline}: one projecting stage per {@link RootType} in the schema, + * each selecting the type’s roots by its source class with blank-node subjects + * excluded ({@link selectByClass} – a blank node has no stable document key), + * extracting with the schema-generated CONSTRUCT, and projecting inside the + * batch; plus the single {@link searchIndexWriter} terminal routing each + * document to the engine writer for its type’s collection. + * + * The consumer supplies only the domain and the deployment shell: the schema, + * which datasets, the engine writer per type, and (optionally) the + * SPARQL/import adapter, provenance and reporting. + * + * ```ts + * const pipeline = searchIndexerPipeline({ + * schema, + * datasets, + * distributionResolver: new ImportResolver(new SparqlDistributionResolver(), { + * ...createQlever({ mode: 'docker', image: 'adfreiburg/qlever:latest' }), + * strategy: 'import', + * }), + * writerFor: (searchType) => new InPlaceRebuild(typesenseClient, searchType), + * }); + * await pipeline.run(); + * ``` + * + * A deployment that needs more – a bespoke root selector (the entry point is a + * domain fact, not a class), per-stage tuning (`batchSize`, `maxConcurrency`), + * non-SPARQL readers, or quad-level plugins – composes {@link searchStages}, + * {@link searchIndexWriter} and {@link Pipeline} directly; this convenience + * owns no capability of its own. + */ +export function searchIndexerPipeline( + options: SearchIndexerPipelineOptions, +): Pipeline { + const { schema, datasets } = options; + return new Pipeline({ + datasetSelector: Array.isArray(datasets) + ? new ManualDatasetSelection(datasets) + : datasets, + distributionResolver: options.distributionResolver, + stages: searchStages({ + schema, + types: [...schema.values()].map((searchType) => ({ + searchType, + rootVariable: 'root', + itemSelector: selectByClass(searchType), + })), + }), + writers: searchIndexWriter({ schema, writerFor: options.writerFor }), + provenanceStore: options.provenanceStore, + pipelineVersion: options.pipelineVersion, + reporter: options.reporter, + }); +} diff --git a/packages/search-pipeline/test/search-indexer-pipeline.integration.test.ts b/packages/search-pipeline/test/search-indexer-pipeline.integration.test.ts new file mode 100644 index 00000000..6242a7a0 --- /dev/null +++ b/packages/search-pipeline/test/search-indexer-pipeline.integration.test.ts @@ -0,0 +1,137 @@ +import { fileURLToPath } from 'node:url'; +import { afterAll, beforeAll, describe, expect, it } from 'vitest'; +import { Dataset, Distribution } from '@lde/dataset'; +import { + ProbedDistributions, + ResolvedDistribution, + type DistributionResolver, + type ProgressReporter, + type Writer, +} from '@lde/pipeline'; +import { + defineSearchType, + searchSchema, + type RootType, + type SearchDocument, +} from '@lde/search'; +import { + startSparqlEndpoint, + teardownSparqlEndpoint, +} from '@lde/local-sparql-endpoint'; +import { searchIndexerPipeline } from '../src/search-indexer-pipeline.js'; + +const SCHEMA = 'https://schema.org/'; + +// The Drapo-shaped schema of the extraction round-trip test, both types root: +// the convenience derives one stage and one collection per root type. +const person = defineSearchType({ + name: 'Person', + class: `${SCHEMA}Person`, + fields: [ + { + name: 'label', + kind: 'text', + path: `<${SCHEMA}name>`, + locales: ['und'], + output: true, + searchable: { weight: 3 }, + }, + ], +}); + +const creativeWork = defineSearchType({ + name: 'CreativeWork', + class: `${SCHEMA}CreativeWork`, + fields: [ + { + name: 'name', + kind: 'text', + path: `<${SCHEMA}name>`, + locales: ['nl', 'en'], + output: true, + searchable: { weight: 5 }, + }, + ], +}); + +const schema = searchSchema(creativeWork, person); + +describe('searchIndexerPipeline end to end', () => { + const port = 3008; + const distribution = Distribution.sparql( + new URL(`http://localhost:${port}/sparql`), + ); + const dataset = new Dataset({ + iri: new URL('http://example.org/dataset/drapo'), + distributions: [distribution], + }); + + // Resolve straight to the local endpoint: probing and importing are the + // resolver’s own concern, tested with @lde/pipeline – here it is a seam. + const resolver: DistributionResolver = { + probe: async () => new ProbedDistributions(dataset, [], null), + resolve: async () => new ResolvedDistribution(distribution, []), + }; + + beforeAll(async () => { + const fixture = fileURLToPath( + new URL('./fixtures/drapo-sample.ttl', import.meta.url), + ); + await startSparqlEndpoint(port, fixture); + }, 60_000); + + afterAll(async () => { + await teardownSparqlEndpoint(); + }); + + it('indexes every root type’s IRI roots into its own committed collection', async () => { + const received = new Map(); + const committed: string[] = []; + const writerFor = (searchType: RootType): Writer => ({ + openRun: async () => ({ + write: async (_dataset, documents) => { + const collection = received.get(searchType.name) ?? []; + received.set(searchType.name, collection); + for await (const document of documents) { + collection.push(document); + } + }, + commit: async () => { + committed.push(searchType.name); + }, + abort: async () => undefined, + }), + }); + + // The pipeline isolates a stage failure per dataset instead of throwing + // (e.g. framing crashing on a blank-node root), so capture it explicitly. + const stageFailures: Error[] = []; + const reporter: ProgressReporter = { + stageFailed: (_stage, error) => { + stageFailures.push(error); + }, + }; + + await searchIndexerPipeline({ + schema, + datasets: [dataset], + distributionResolver: resolver, + writerFor, + reporter, + }).run(); + + expect(stageFailures).toEqual([]); + // Both roots per type – and not the fixture’s blank-node CreativeWork, + // which selectByClass must exclude (it has no stable document key). + expect( + received + .get('CreativeWork') + ?.map((document) => document.id) + .sort(), + ).toEqual(['https://ex/cw/1', 'https://ex/cw/2']); + expect(received.get('Person')?.map((document) => document.id)).toEqual([ + 'https://ex/p/1', + ]); + expect(committed.sort()).toEqual(['CreativeWork', 'Person']); + }); +}); diff --git a/packages/search-pipeline/test/search-indexer-pipeline.test.ts b/packages/search-pipeline/test/search-indexer-pipeline.test.ts new file mode 100644 index 00000000..6ca058a6 --- /dev/null +++ b/packages/search-pipeline/test/search-indexer-pipeline.test.ts @@ -0,0 +1,96 @@ +import { describe, expect, it, vi } from 'vitest'; +import { Dataset } from '@lde/dataset'; +import { + ManualDatasetSelection, + Pipeline, + type ProvenanceStore, + type Writer, +} from '@lde/pipeline'; +import { searchSchema, type RootType, type SearchDocument } from '@lde/search'; +import { searchIndexerPipeline } from '../src/search-indexer-pipeline.js'; + +const DATASET = 'https://example.org/Dataset'; +const ORGANIZATION = 'https://example.org/Organization'; + +const schema = searchSchema( + { + name: 'Dataset', + class: DATASET, + fields: [ + { + name: 'title', + kind: 'keyword', + path: '', + output: true, + }, + ], + }, + { + name: 'Organization', + class: ORGANIZATION, + fields: [ + { + name: 'name', + kind: 'keyword', + path: '', + output: true, + }, + ], + }, +); + +const engineWriter: Writer = { + openRun: async () => ({ + write: async () => undefined, + commit: async () => undefined, + abort: async () => undefined, + }), +}; + +describe('searchIndexerPipeline', () => { + it('wires a Pipeline from a dataset list, one engine writer per root type', () => { + const writerFor = vi.fn((_searchType: RootType) => engineWriter); + const pipeline = searchIndexerPipeline({ + schema, + datasets: [ + new Dataset({ + iri: new URL('http://example.org/dataset/1'), + distributions: [], + }), + ], + writerFor, + }); + + expect(pipeline).toBeInstanceOf(Pipeline); + // One engine writer per root type in the schema, built eagerly. + expect(writerFor).toHaveBeenCalledTimes(2); + expect(writerFor.mock.calls.map(([type]) => type)).toEqual([ + schema.get(DATASET), + schema.get(ORGANIZATION), + ]); + }); + + it('accepts a DatasetSelector for dynamic selection', () => { + const pipeline = searchIndexerPipeline({ + schema, + datasets: new ManualDatasetSelection([]), + writerFor: () => engineWriter, + }); + expect(pipeline).toBeInstanceOf(Pipeline); + }); + + it('rejects a provenance store without a pipeline version', () => { + const provenanceStore: ProvenanceStore = { + get: async () => null, + set: async () => undefined, + }; + expect(() => + searchIndexerPipeline({ + schema, + datasets: [], + writerFor: () => engineWriter, + provenanceStore, + }), + ).toThrow(/pipelineVersion is required/); + }); +}); diff --git a/packages/sparql-importer/README.md b/packages/sparql-importer/README.md index cf1de857..9764607d 100644 --- a/packages/sparql-importer/README.md +++ b/packages/sparql-importer/README.md @@ -4,3 +4,14 @@ Import RDF data dumps to a local SPARQL endpoint. This is useful when: - the dataset does not offer a SPARQL endpoint itself - or when it does but the endpoint is too slow or unreliable, particularly for more complex SPARQL queries. + +This package is **interface-only**: it defines the `Importer` contract and its +result types (`ImportSuccessful`, `ImportFailed`, `NotSupported`) that pipeline +components program against. It ships no importer of its own. + +The concrete adapter is [`@lde/sparql-qlever`](../sparql-qlever), whose +`Importer` downloads a distribution and builds a +[QLever](https://github.com/ad-freiburg/qlever) index from it, and whose +`createQlever` pairs it with a matching +[`@lde/sparql-server`](../sparql-server) `Server` for use in an +[`@lde/pipeline`](../pipeline) `ImportResolver`. diff --git a/packages/sparql-server/README.md b/packages/sparql-server/README.md index e92d286b..538b1202 100644 --- a/packages/sparql-server/README.md +++ b/packages/sparql-server/README.md @@ -1,3 +1,13 @@ # SPARQL Server Start, stop and control SPARQL servers with a simple API. + +This package is **interface-only**: it defines the `SparqlServer` contract +(`start()`, `stop()`, `queryEndpoint`) that pipeline components program +against. It ships no server of its own. + +The concrete adapter is [`@lde/sparql-qlever`](../sparql-qlever), whose +`Server` runs a [QLever](https://github.com/ad-freiburg/qlever) instance +(Docker or native) and whose `createQlever` pairs it with a matching +[`@lde/sparql-importer`](../sparql-importer) `Importer` for use in an +[`@lde/pipeline`](../pipeline) `ImportResolver`.