diff --git a/AGENTS.md b/AGENTS.md index 6b115f06..ccbec12d 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -86,7 +86,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), `search-pipeline` extraction round-trip (3007), `search-pipeline` searchIndexerPipeline end-to-end (3008), `search-pipeline` registry-sourced extraction (3009), `search-pipeline` joins end-to-end (3010), `search-pipeline` keyed roots (3011), `search-pipeline` nested fan-out (3012), `pipeline-void` per-property partitions (3013), `pipeline-void` subject URI spaces (3014), `pipeline-void` distinct counts (3016), `sparql-qlever` remote mode (3015 for QLever, 7070 for the task agent) +- 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), `search-pipeline` extraction round-trip (3007), `search-pipeline` searchIndexerPipeline end-to-end (3008), `search-pipeline` registry-sourced extraction (3009), `search-pipeline` joins end-to-end (3010), `search-pipeline` keyed roots (3011), `search-pipeline` nested fan-out (3012), `pipeline-void` per-property partitions (3013), `pipeline-void` subject URI spaces (3014), `pipeline-void` distinct counts (3016), `sparql-qlever` remote mode (3015 for QLever, 7070 for the task agent), `pipeline-void` scalar counts (3017) ### Key Dependencies diff --git a/docs/reference/pipeline.md b/docs/reference/pipeline.md index ffdfd726..118167e1 100644 --- a/docs/reference/pipeline.md +++ b/docs/reference/pipeline.md @@ -214,7 +214,7 @@ interface ItemSelector { } ``` -The distribution is received at run time, so selectors don't need the endpoint URL at construction time. The `batchSize` parameter is set by the stage; `options` carries the per-dataset [timeout policy](#timeout-policies). Use `SparqlItemSelector` for SPARQL-based selection with automatic pagination: +The distribution is received at run time, so selectors don't need the endpoint URL at construction time. The `batchSize` parameter is set by the stage; `options` carries the per-dataset [timeout policy](#timeout-policies) and the same [dataset bindings](#dataset-bindings) a reader gets. Use `SparqlItemSelector` for SPARQL-based selection with automatic pagination: ```typescript new SparqlItemSelector({ @@ -272,7 +272,8 @@ The query is templated per dataset and distribution, in order: 1. `#subjectFilter#` is replaced with `distribution.subjectFilter` (the empty string when unset). 2. When the distribution has a `namedGraph`, a `FROM ` clause is injected for it. -3. Every literal `?dataset` occurrence is replaced with the dataset’s IRI. This includes a genuine `?dataset` variable, which gets rewritten too – don’t name an ordinary query variable `?dataset`. +3. Per-item (selector) bindings, if any, are injected as a `VALUES` clause into the innermost sub-SELECT. +4. [Dataset bindings](#dataset-bindings) the query references are injected as a `VALUES` clause at the outermost scope. A `Stage` passes these. When calling `read()` yourself, pass them as `datasetBindings`, for example `predefinedDatasetBindings(dataset)`, or nothing is bound. Transient failures – network errors and HTTP 502/503/504 – are retried; `retries` (default: 3) sets how many times. `SparqlConstructReaderOptions.fetcher` injects a custom `SparqlEndpointFetcher`, intended for tests; the reader then uses it as-is, so the per-request budget from the [timeout policy](#timeout-policies) is not enforced (the policy’s hooks still fire, but its own `timeout` governs). @@ -285,7 +286,7 @@ const reader = new SparqlConstructReader({ }); ``` -SPARQL CONSTRUCT queries can produce duplicate triples – for example, constant triples (like `?dataset a edm:ProvidedCHO`) are emitted for every solution row. Enable `deduplicate` to remove duplicates inline on the stream using a string-based identity set (inspired by [Comunica's `distinctConstruct`](https://comunica.dev/docs/query/advanced/context/#14--distinct-construct)): +SPARQL CONSTRUCT queries can produce duplicate triples – for example, constant triples (like `$this a edm:ProvidedCHO`) are emitted for every solution row. Enable `deduplicate` to remove duplicates inline on the stream using a string-based identity set (inspired by [Comunica's `distinctConstruct`](https://comunica.dev/docs/query/advanced/context/#14--distinct-construct)): ```typescript const reader = new SparqlConstructReader({ @@ -294,10 +295,51 @@ const reader = new SparqlConstructReader({ }); ``` -The dedup set is scoped to each `read()` call, so memory stays bounded to the number of unique quads per batch. A standalone `deduplicateQuads()` function is also exported for use outside the reader. +This happens in the client: the endpoint still computes and sends every duplicate, and they still count towards a `LIMIT` in the query. The dedup set is scoped to each `read()` call, so memory stays bounded to the number of unique quads per batch, and duplicates across batches or readers remain. A standalone `deduplicateQuads()` function is also exported for use outside the reader. Logic that is hard to express in pure SPARQL – cleaning up messy date notations, converting locale-specific dates to ISO 8601 – can be attached to a reader as a **quad transform**, a plain function that post-processes the reader’s output. See the how-to guide [Extend a stage with a quad transform](../guide/extend-a-stage-with-a-quad-transform). +### Dataset bindings + +Beyond `?dataset`, a reader or `SparqlItemSelector` can reach a handful of DCAT properties from the `Dataset` in hand, under reserved variable names. Don't reuse these for an unrelated query variable. + +| Variable | From | Multi-valued? | +| ----------------------- | -------------------------------------------- | ---------------------------- | +| `?dataset` | `dataset.iri` | No | +| `?datasetPublisher` | `dataset.publisher.iri` | No | +| `?datasetPublisherName` | `dataset.publisher.name` (language → string) | Yes, one row per translation | +| `?datasetLicense` | `dataset.license` | No | +| `?datasetLanguage` | `dataset.language` | Yes, one row per language | + +A property absent for a given dataset (no `publisher`, no `license`) leaves its variable unbound rather than `UNDEF`. + +Use these like any other variable: the pipeline binds the ones your query references, and a query that references none runs unchanged. Bindings don't reach inside a sub-SELECT, so a variable used only there stays unbound. + +A multi-valued variable takes each of its values: + +- Used only in the `CONSTRUCT` template, each value produces its own triple, independently of other multi-valued variables. To get every combination of two of them, use both in one template triple. +- Used anywhere else, such as in a pattern, `FILTER` or `BIND`, the query runs for every combination of its values with those of other multi-valued variables it references. + +Either way, the query's other triples repeat once per value. Enable [`deduplicate`](#reader) if that matters. In a selector, a multi-valued variable repeats the selected items, so use `SELECT DISTINCT`. + +For a value this library can't know implement `DatasetBindingsProvider` and pass it to `Stage`: + +```typescript +const bindingsProvider: DatasetBindingsProvider = { + bindings(dataset) { + return [{ type: namedNode(lookupLocalType(dataset)) }]; + }, +}; + +new Stage({ + name: 'enrich', + readers: reader, + bindingsProvider, +}); +``` + +A provider's rows go to every reader and, when configured, the item selector. They follow the same rules, with several rows acting as one multi-valued variable. Values that must appear in one solution together, such as a language and a name in that language, belong in the same row. A variable the rows bind replaces the predefined binding of that name. It may return a `Promise`. + ### Timeout policies By default, every SPARQL request uses the same 5-minute budget. When a pipeline runs against many third-party endpoints, that fixed budget can cost ~80 minutes on a single dataset whose endpoint times out repeatedly on heavy queries – light stages on the same endpoint then sit behind the heavy ones that will never succeed. diff --git a/packages/pipeline-void/test/scalarCounts.integration.test.ts b/packages/pipeline-void/test/scalarCounts.integration.test.ts new file mode 100644 index 00000000..20250ec9 --- /dev/null +++ b/packages/pipeline-void/test/scalarCounts.integration.test.ts @@ -0,0 +1,95 @@ +import { + countObjectLiterals, + countObjectUris, + countProperties, + countSubjects, + countTriples, + Stage, +} from '../src/index.js'; +import { Dataset, Distribution } from '@lde/dataset'; +import type { DatasetWriter } from '@lde/pipeline'; +import { + startSparqlEndpoint, + teardownSparqlEndpoint, +} from '@lde/local-sparql-endpoint'; +import type { Quad } from '@rdfjs/types'; +import { fetch as undiciFetch, setGlobalDispatcher, Agent } from 'undici'; +import { fileURLToPath } from 'node:url'; +import { describe, it, expect, beforeAll, afterAll } from 'vitest'; + +// Pin fetch + dispatcher to one undici copy – see +// namespaceNormalization.integration.test.ts for the rationale. +setGlobalDispatcher(new Agent()); +globalThis.fetch = undiciFetch as unknown as typeof globalThis.fetch; + +const fixture = fileURLToPath( + new URL('./fixtures/mixedNamespaces.ttl', import.meta.url), +); + +const VOID = 'http://rdfs.org/ns/void#'; +const VOID_EXT = 'http://ldf.fi/void-ext#'; + +const dataset = new Dataset({ + iri: new URL('http://example.org/dataset'), + distributions: [], +}); + +// These queries' WHERE clause is a bare sub-SELECT, which the dataset VALUES +// injection must keep syntactically valid. +describe('scalar count stages (end to end)', () => { + // Unique across the repo's integration suites – nx runs packages' tests + // concurrently, so a shared port is a flaky EADDRINUSE. + const port = 3017; + const distribution = Distribution.sparql( + new URL(`http://localhost:${port}/sparql`), + ); + + beforeAll(async () => { + await startSparqlEndpoint(port, fixture); + }, 60_000); + + afterAll(async () => { + await teardownSparqlEndpoint(); + }, 30_000); + + // Expected counts follow from mixedNamespaces.ttl. + it.each<[string, () => Promise, string, string]>([ + ['countSubjects', countSubjects, `${VOID}distinctSubjects`, '5'], + ['countTriples', countTriples, `${VOID}triples`, '11'], + ['countProperties', countProperties, `${VOID}properties`, '5'], + [ + 'countObjectLiterals', + countObjectLiterals, + `${VOID_EXT}distinctLiterals`, + '4', + ], + [ + 'countObjectUris', + countObjectUris, + `${VOID_EXT}distinctIRIReferenceObjects`, + '4', + ], + ])( + '%s', + async (_name, createStage, predicate, expected) => { + const quads: Quad[] = []; + const writer: DatasetWriter = { + async write(_dataset, data) { + for await (const q of data) quads.push(q); + }, + }; + + await (await createStage()).run(dataset, distribution, writer); + + const counts = quads + .filter( + (q) => + q.subject.value === dataset.iri.toString() && + q.predicate.value === predicate, + ) + .map((q) => q.object.value); + expect(counts).toEqual([expected]); + }, + 30_000, + ); +}); diff --git a/packages/pipeline/src/sparql/datasetBindings.ts b/packages/pipeline/src/sparql/datasetBindings.ts new file mode 100644 index 00000000..1dca6a7c --- /dev/null +++ b/packages/pipeline/src/sparql/datasetBindings.ts @@ -0,0 +1,81 @@ +import { Dataset, Distribution } from '@lde/dataset'; +import { DataFactory } from 'n3'; +import type { VariableBindings } from './reader.js'; + +const { namedNode, literal } = DataFactory; + +/** Reserved variable names – see {@link predefinedDatasetBindings}. */ +export const DATASET_BINDING_VARIABLES = [ + 'dataset', + 'datasetPublisher', + 'datasetPublisherName', + 'datasetLicense', + 'datasetLanguage', +] as const; + +/** + * Dataset-level bindings as groups of rows. Rows in one group are alternatives + * for the same variables; different groups say nothing about each other, so a + * query only pays for a combination of them if it needs one – see + * {@link injectDatasetBindings}. A group may be empty. + */ +export type DatasetBindings = VariableBindings[][]; + +/** + * Always-on bindings derived from a {@link Dataset} – see + * {@link DATASET_BINDING_VARIABLES}. The single-valued properties share one + * group; `datasetLanguage` and `datasetPublisherName` each form their own, + * because a language and a publisher name say nothing about each other. + */ +export function predefinedDatasetBindings(dataset: Dataset): DatasetBindings { + const base: VariableBindings = { dataset: namedNode(dataset.iri.toString()) }; + + if (dataset.publisher) { + base.datasetPublisher = namedNode(dataset.publisher.iri.toString()); + } + if (dataset.license) { + base.datasetLicense = namedNode(dataset.license.toString()); + } + + const languages = dataset.language.map((language) => ({ + datasetLanguage: literal(language), + })); + const publisherNames = Object.entries(dataset.publisher?.name ?? {}).map( + ([language, name]) => ({ datasetPublisherName: literal(name, language) }), + ); + + return [[base], languages, publisherNames]; +} + +/** + * The {@link predefinedDatasetBindings} plus a provider's rows as one more + * group. A variable the rows bind replaces the predefined binding of that name. + */ +export function datasetBindingsFor( + dataset: Dataset, + customRows: VariableBindings[], +): DatasetBindings { + const customVariables = new Set(customRows.flatMap(Object.keys)); + const predefined = predefinedDatasetBindings(dataset).map((rows) => + rows.map((row) => + Object.fromEntries( + Object.entries(row).filter(([name]) => !customVariables.has(name)), + ), + ), + ); + + return [...predefined, customRows]; +} + +/** + * Extension point for per-dataset bindings this library can't know about + * (e.g. a caller-local `rdf:type` read from a side file). Mirrors + * {@link Reader} and {@link ItemSelector}. The returned rows form one group: + * return values that must appear in one solution together in the same row. + */ +export interface DatasetBindingsProvider { + bindings( + dataset: Dataset, + distribution: Distribution, + ): VariableBindings[] | Promise; +} diff --git a/packages/pipeline/src/sparql/index.ts b/packages/pipeline/src/sparql/index.ts index eb37d89e..b4f61b41 100644 --- a/packages/pipeline/src/sparql/index.ts +++ b/packages/pipeline/src/sparql/index.ts @@ -14,7 +14,15 @@ export { type SparqlItemSelectorOptions, } from './selector.js'; -export { injectValues } from './values.js'; +export { injectValues, injectDatasetBindings } from './values.js'; + +export { + predefinedDatasetBindings, + datasetBindingsFor, + DATASET_BINDING_VARIABLES, + type DatasetBindings, + type DatasetBindingsProvider, +} from './datasetBindings.js'; export { withDefaultGraph } from './graph.js'; diff --git a/packages/pipeline/src/sparql/reader.ts b/packages/pipeline/src/sparql/reader.ts index 32a263d2..ecf182a1 100644 --- a/packages/pipeline/src/sparql/reader.ts +++ b/packages/pipeline/src/sparql/reader.ts @@ -1,6 +1,6 @@ -import { Dataset, Distribution, assertSafeIri } from '@lde/dataset'; +import { Dataset, Distribution } from '@lde/dataset'; import { SparqlEndpointFetcher } from 'fetch-sparql-endpoint'; -import type { NamedNode, Quad } from '@rdfjs/types'; +import type { NamedNode, Literal, Quad } from '@rdfjs/types'; import { readFile } from 'node:fs/promises'; import { resolve } from 'node:path'; import { Transform } from 'node:stream'; @@ -12,7 +12,8 @@ import isNetworkError from 'is-network-error'; import pRetry from 'p-retry'; import { quadToStringQuad } from 'rdf-string'; import { withDefaultGraph } from './graph.js'; -import { injectValues } from './values.js'; +import { injectDatasetBindings, injectValues } from './values.js'; +import type { DatasetBindings } from './datasetBindings.js'; import { ConstantTimeoutPolicy, type TimeoutOutcome, @@ -33,16 +34,24 @@ export class NotSupported { constructor(public readonly message: string) {} } -/** A single row of variable bindings (variable name → NamedNode). */ -export type VariableBindings = Record; +/** A single row of variable bindings (variable name → NamedNode or Literal). */ +export type VariableBindings = Record; export interface ReadOptions { /** - * Variable bindings to inject as a VALUES clause into the query. - * When non-empty, a VALUES block is prepended to the WHERE clause. + * Per-item variable bindings (typically an {@link ItemSelector}'s rows). + * Injected at the innermost sub-SELECT – see {@link injectValues}. */ bindings?: VariableBindings[]; + /** + * Dataset-level bindings, injected at the outermost scope for the variables + * the query references – see {@link injectDatasetBindings}. A {@link Stage} + * passes these. Without them nothing is bound, not even `?dataset`: outside a + * stage, pass {@link predefinedDatasetBindings} to bind the DCAT variables. + */ + datasetBindings?: DatasetBindings; + /** * Per-call {@link TimeoutPolicy}. When supplied, the reader calls * {@link TimeoutPolicy.beforeRequest} once per attempt (including @@ -113,12 +122,18 @@ export interface SparqlConstructReaderOptions { * Deduplicate triples in the CONSTRUCT output stream. * * SPARQL CONSTRUCT queries can produce duplicate triples — for example, - * constant triples (like `?dataset a edm:ProvidedCHO`) are emitted for - * every solution row. When enabled, a streaming identity filter removes - * duplicates inline without buffering. + * constant triples (like `$this a edm:ProvidedCHO`) are emitted for every + * solution row; a multi-valued dataset binding (`?datasetLanguage`, + * `?datasetPublisherName`) that the query references does the same. When + * enabled, a streaming identity filter removes duplicates inline without + * buffering. + * + * This happens in the client: the endpoint still computes and sends every + * duplicate, and they still count towards a `LIMIT` in the query. * * The dedup set is scoped to each {@link read} call, so memory stays - * bounded to the number of unique quads per call (typically one batch). + * bounded to the number of unique quads per call (typically one batch), and + * duplicates across batches or readers remain. * * Inspired by Comunica's `distinctConstruct` context option. * @@ -137,14 +152,20 @@ export interface SparqlConstructReaderOptions { * Template substitution (applied in order): * 1. `#subjectFilter#` — replaced with `distribution.subjectFilter` (deferred to read) * 2. `FROM ` — set via `withDefaultGraph` if the distribution has a named graph - * 3. `?dataset` — replaced with the dataset IRI (string substitution on the serialised query) + * 3. Per-item (selector) bindings — injected at the innermost sub-SELECT + * (see {@link injectValues}). + * 4. Dataset-level bindings passed in {@link ReadOptions.datasetBindings} — + * injected at the outermost scope for the variables the query references + * (see {@link injectDatasetBindings}). * * @example * ```typescript * const reader = new SparqlConstructReader({ * query: 'CONSTRUCT { ?dataset ?p ?o } WHERE { ?s ?p ?o }', * }); - * const result = await reader.read(dataset, distribution); + * const result = await reader.read(dataset, distribution, { + * datasetBindings: predefinedDatasetBindings(dataset), + * }); * if (result instanceof NotSupported) { * console.log(result.message); * } else { @@ -184,13 +205,13 @@ export class SparqlConstructReader implements Reader { * Read quads by executing the SPARQL CONSTRUCT query against the * distribution's endpoint. * - * @param dataset The dataset to read from. + * @param _dataset The dataset to read from. * @param distribution The distribution providing the SPARQL endpoint. - * @param options Optional read options (bindings). + * @param options Optional read options (bindings, dataset bindings). * @returns AsyncIterable stream of results. */ async read( - dataset: Dataset, + _dataset: Dataset, distribution: Distribution, options?: ReadOptions, ): Promise> { @@ -215,14 +236,16 @@ export class SparqlConstructReader implements Reader { withDefaultGraph(ast, distribution.namedGraph); } - const bindings = options?.bindings; - if (bindings !== undefined && bindings.length > 0) { - ast = injectValues(ast, bindings); + const itemBindings = options?.bindings ?? []; + if (itemBindings.length > 0) { + ast = injectValues(ast, itemBindings); + } + + if (options?.datasetBindings) { + ast = injectDatasetBindings(ast, options.datasetBindings); } - let query = this.generator.generate(ast); - assertSafeIri(dataset.iri.toString()); - query = query.replaceAll('?dataset', `<${dataset.iri}>`); + const query = this.generator.generate(ast); const policy = options?.timeout ?? defaultTimeoutPolicy; diff --git a/packages/pipeline/src/sparql/selector.ts b/packages/pipeline/src/sparql/selector.ts index bd561d50..d4dbf347 100644 --- a/packages/pipeline/src/sparql/selector.ts +++ b/packages/pipeline/src/sparql/selector.ts @@ -10,6 +10,7 @@ import { } from '@traqula/rules-sparql-1-1'; import type { ItemSelector, SelectOptions } from '../stage.js'; import { withDefaultGraph } from './graph.js'; +import { injectDatasetBindings } from './values.js'; import type { VariableBindings } from './reader.js'; import { ConstantTimeoutPolicy, @@ -136,10 +137,13 @@ 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); + let query = structuredClone(this.parsed); if (distribution.namedGraph) { withDefaultGraph(query, distribution.namedGraph); } + if (options?.datasetBindings) { + query = injectDatasetBindings(query, options.datasetBindings); + } const policy = options?.timeout ?? defaultTimeoutPolicy; let offset = 0; let totalFetched = 0; diff --git a/packages/pipeline/src/sparql/values.ts b/packages/pipeline/src/sparql/values.ts index 4327e038..0bacf50d 100644 --- a/packages/pipeline/src/sparql/values.ts +++ b/packages/pipeline/src/sparql/values.ts @@ -6,11 +6,39 @@ import { type PatternValues, type QueryConstruct, type QuerySelect, + type TermIri, + type TermLiteral, + type TermVariable, type ValuePatternRow, } from '@traqula/rules-sparql-1-1'; +import type { NamedNode, Literal } from '@rdfjs/types'; +import { termToString } from 'rdf-string'; +import type { DatasetBindings } from './datasetBindings.js'; import type { VariableBindings } from './reader.js'; +import { assertSafeIri } from '@lde/dataset'; const F = new AstFactory(); +const XSD_STRING = 'http://www.w3.org/2001/XMLSchema#string'; + +function termToAstTerm(node: NamedNode | Literal): TermIri | TermLiteral { + if (node.termType === 'Literal') { + if (node.language) return F.termLiteral(F.gen(), node.value, node.language); + if (node.datatype.value === XSD_STRING) { + return F.termLiteral(F.gen(), node.value); + } + + assertSafeIri(node.datatype.value); + return F.termLiteral( + F.gen(), + node.value, + F.termNamed(F.gen(), node.datatype.value), + ); + } + + // The generator writes IRIs into the query unescaped. + assertSafeIri(node.value); + return F.termNamed(F.gen(), node.value); +} /** * Find the first SubSelect within a list of patterns, looking through @@ -133,39 +161,185 @@ function bindsAllVariables( return valuesPattern.variables.every((variable) => bound.has(variable.value)); } +function buildValuesPattern(bindings: VariableBindings[]): PatternValues { + const variableNames = [...new Set(bindings.flatMap(Object.keys))]; + + const variables = variableNames.map((name) => F.termVariable(name, F.gen())); + + const values: ValuePatternRow[] = bindings.map((row) => + Object.fromEntries( + Object.entries(row).map(([name, node]) => [name, termToAstTerm(node)]), + ), + ); + + return F.patternValues(variables, values, F.gen()); +} + /** - * Inject a VALUES clause into a parsed CONSTRUCT query for the given binding rows. - * - * Each row's keys become SPARQL variables; NamedNode values become IRIs in the - * VALUES block. The VALUES clause is injected into the innermost subquery so - * that SPARQL engines can constrain scans early. When that WHERE clause is a - * single UNION whose branches all bind the VALUES variables in a required - * triple pattern, the VALUES goes into each branch instead: engines such as - * QLever plan and evaluate each branch without the outer VALUES otherwise. + * Injects a VALUES clause at the innermost sub-SELECT, for per-item (selector) + * bindings only, so that SPARQL engines can constrain scans early. When that + * WHERE clause is a single UNION whose branches all bind the VALUES variables + * in a required triple pattern, the VALUES goes into each branch instead: + * engines such as QLever plan and evaluate each branch without the outer + * VALUES otherwise. * - * The caller owns parsing and stringifying; this function operates on the AST. + * Dataset-level bindings must use {@link injectDatasetBindings} instead – a + * sub-SELECT doesn't inherit outer bindings, so injecting them here would + * leave an outer `BIND`/`FILTER` referencing them unbound. */ -export function injectValues( - query: QueryConstruct, +export function injectValues( + query: Q, bindings: VariableBindings[], -): QueryConstruct { - const variableNames = bindings.length > 0 ? Object.keys(bindings[0]) : []; +): Q { + return { + ...query, + where: injectIntoInnermost(query.where, buildValuesPattern(bindings)), + }; +} - const variables = variableNames.map((name) => F.termVariable(name, F.gen())); +/** + * The variables in an AST node that a binding at the outermost scope can reach. + * Of a sub-SELECT only the projection counts, because the rest is out of scope. + */ +function visibleVariables(node: unknown): string[] { + if (Array.isArray(node)) return node.flatMap(visibleVariables); + if (typeof node !== 'object' || node === null) return []; + if (F.isTermVariable(node)) return [(node as TermVariable).value]; + if (F.isQuerySelect(node)) return projectedVariables(node as QuerySelect); - const values: ValuePatternRow[] = bindings.map((row) => - Object.fromEntries( - Object.entries(row).map(([name, node]) => [ - name, - F.termNamed(F.gen(), node.value), - ]), + return visibleVariables(Object.values(node)); +} + +function projectedVariables(subSelect: QuerySelect): string[] { + return subSelect.variables.flatMap((projected) => { + if (F.isWildcard(projected)) return visibleVariables(subSelect.where); + if (F.isTermVariable(projected)) return [projected.value]; + + return [projected.variable.value]; + }); +} + +/** A key that is equal for rows binding the same variables to the same terms. */ +function rowKey(row: VariableBindings): string { + return JSON.stringify( + Object.entries(row) + .map(([name, term]) => [name, termToString(term)]) + .sort(([a], [b]) => a.localeCompare(b)), + ); +} + +/** Narrows rows to the given variables, dropping rows that become equal or empty. */ +function narrowRows( + rows: VariableBindings[], + variables: Set, +): VariableBindings[] { + const narrowed = rows + .map((row) => + Object.fromEntries( + Object.entries(row).filter(([name]) => variables.has(name)), + ), + ) + .filter((row) => Object.keys(row).length > 0); + return [...new Map(narrowed.map((row) => [rowKey(row), row])).values()]; +} + +function overlaps(a: Set, b: Set): boolean { + return [...a].some((variable) => b.has(variable)); +} + +/** + * Whether a multi-row group can get its own `UNION` branch: the query only + * outputs its variables, in template triples without another group's variables. + */ +function canBranch( + rows: VariableBindings[], + multiRowGroups: VariableBindings[][], + usedOutsideTemplate: Set, + templateTriples: Set[], +): boolean { + const own = new Set(rows.flatMap(Object.keys)); + const others = new Set( + multiRowGroups + .filter((group) => group !== rows) + .flat() + .flatMap(Object.keys), + ); + + return ( + !overlaps(own, usedOutsideTemplate) && + !templateTriples.some( + (variables) => overlaps(variables, own) && overlaps(variables, others), + ) + ); +} + +/** One group needs no `UNION`; several each get their own branch. */ +function buildBranchPatterns(branches: VariableBindings[][]): Pattern[] { + if (branches.length === 0) return []; + if (branches.length === 1) return [buildValuesPattern(branches[0])]; + + return [ + F.patternUnion( + branches.map((rows) => + F.patternGroup([buildValuesPattern(rows)], F.gen()), + ), + F.gen(), ), + ]; +} + +/** + * Injects dataset-level bindings at the outermost WHERE scope, for the variables + * the query references. A query that references none is returned unchanged. + * + * Multi-row groups the query only outputs go in separate `UNION` branches, so + * L languages and N names make L + N rows instead of L × N. Other multi-row + * groups are joined: in another group's branch their variables would be unbound, + * making patterns on them match too much and filters on them remove solutions. + */ +export function injectDatasetBindings( + query: Q, + bindings: DatasetBindings, +): Q { + const { template, ...clauses } = query as Q & Partial; + const templateTriples = (template?.triples ?? []).map( + (triple) => new Set(visibleVariables(triple)), ); + const usedOutsideTemplate = new Set(visibleVariables(Object.values(clauses))); + const referenced = new Set([ + ...usedOutsideTemplate, + ...templateTriples.flatMap((variables) => [...variables]), + ]); + + const groups = bindings + .map((rows) => narrowRows(rows, referenced)) + .filter((rows) => rows.length > 0); + if (groups.length === 0) return query; - const valuesPattern = F.patternValues(variables, values, F.gen()); + const singleRows = groups.filter((rows) => rows.length === 1).flat(); + const multiRowGroups = groups.filter((rows) => rows.length > 1); + + const branches = multiRowGroups.filter((rows) => + canBranch(rows, multiRowGroups, usedOutsideTemplate, templateTriples), + ); + const joined = multiRowGroups.filter((rows) => !branches.includes(rows)); + + const injected = [ + ...(singleRows.length > 0 + ? [buildValuesPattern([Object.assign({}, ...singleRows)])] + : []), + ...joined.map((rows) => buildValuesPattern(rows)), + ...buildBranchPatterns(branches), + ]; + + // A bare sub-SELECT (`WHERE { SELECT … }`) is only valid as a group's sole + // content, so it needs its own braces once VALUES sits beside it. + const patterns = query.where.patterns.map((pattern) => + F.isQuerySelect(pattern) ? F.patternGroup([pattern], F.gen()) : pattern, + ); return { ...query, - where: injectIntoInnermost(query.where, valuesPattern), + where: F.patternGroup([...injected, ...patterns], F.gen()), }; } diff --git a/packages/pipeline/src/stage.ts b/packages/pipeline/src/stage.ts index 748b1fe1..dfd60419 100644 --- a/packages/pipeline/src/stage.ts +++ b/packages/pipeline/src/stage.ts @@ -3,6 +3,11 @@ import type { Quad } from '@rdfjs/types'; import type { Reader, VariableBindings } from './sparql/reader.js'; import { NotSupported } from './sparql/reader.js'; import type { TimeoutPolicy } from './sparql/timeoutPolicy.js'; +import { + datasetBindingsFor, + type DatasetBindings, + type DatasetBindingsProvider, +} from './sparql/datasetBindings.js'; import { batch } from './batch.js'; import type { Validator } from './validator.js'; import type { DatasetWriter } from './writer/writer.js'; @@ -178,6 +183,13 @@ export interface StageOptions { /** What to do when a batch fails validation. @default 'write' */ onInvalid?: 'write' | 'skip' | 'halt'; }; + /** + * Supplies additional per-dataset bindings beyond the built-in DCAT ones + * (see {@link DATASET_BINDING_VARIABLES}), for values this library can't + * know about (e.g. a caller-local `edm:type`). Handed to both the item + * selector and every reader. + */ + bindingsProvider?: DatasetBindingsProvider; } export interface RunOptions { @@ -195,6 +207,12 @@ export interface RunOptions { export interface SelectOptions { /** Per-call timeout policy. */ timeout?: TimeoutPolicy; + /** + * Dataset-level bindings so an {@link ItemSelector} can inject the same + * bindings a reader would, e.g. to filter selected items by + * `?datasetLicense` – see {@link injectDatasetBindings}. + */ + datasetBindings?: DatasetBindings; } export class Stage { @@ -213,6 +231,7 @@ export class Stage { private readonly project?: BatchTransform; private readonly queueCapacity?: number; private readonly sourceFor?: StageOptions['sourceFor']; + private readonly bindingsProvider?: DatasetBindingsProvider; constructor(options: StageOptions) { if (options.project && !options.itemSelector) { @@ -247,6 +266,7 @@ export class Stage { this.queueCapacity = options.queueCapacity; this.sourceFor = options.sourceFor; this.sourcesOwnData = options.sourceFor !== undefined; + this.bindingsProvider = options.bindingsProvider; } /** The validator for this stage, if configured. */ @@ -266,19 +286,33 @@ export class Stage { const distribution = this.sourceFor ? this.sourceFor(dataset, resolved) : resolved; + const datasetBindings = datasetBindingsFor( + dataset, + this.bindingsProvider + ? await this.bindingsProvider.bindings(dataset, distribution) + : [], + ); + if (this.itemSelector) { return this.runWithSelector( this.itemSelector.select(distribution, this.batchSize, { timeout, + datasetBindings, }), dataset, distribution, writer, options, + datasetBindings, ); } - const streams = await this.readAll(dataset, distribution, timeout); + const streams = await this.readAll( + dataset, + distribution, + timeout, + datasetBindings, + ); if (streams instanceof NotSupported) { return streams; } @@ -355,7 +389,8 @@ export class Stage { dataset: Dataset, distribution: Distribution, writer: DatasetWriter, - options?: RunOptions, + options: RunOptions | undefined, + datasetBindings: DatasetBindings, ): Promise { // Peek the first batch to detect an empty selector before starting the // writer (important because e.g. SparqlUpdateWriter does CLEAR GRAPH). @@ -430,6 +465,7 @@ export class Stage { this.readers.map(async ({ reader, transforms }) => { const result = await reader.read(dataset, distribution, { bindings, + datasetBindings, timeout: options?.timeout, }); if (result instanceof NotSupported) return []; @@ -556,10 +592,12 @@ export class Stage { dataset: Dataset, distribution: Distribution, timeout: TimeoutPolicy | undefined, + datasetBindings: DatasetBindings, ): Promise[] | NotSupported> { const results = await Promise.all( this.readers.map(async ({ reader, transforms }) => { const result = await reader.read(dataset, distribution, { + datasetBindings, timeout, }); if (result instanceof NotSupported) return result; diff --git a/packages/pipeline/test/sparql/datasetBindings.test.ts b/packages/pipeline/test/sparql/datasetBindings.test.ts new file mode 100644 index 00000000..3bd21712 --- /dev/null +++ b/packages/pipeline/test/sparql/datasetBindings.test.ts @@ -0,0 +1,99 @@ +import { describe, it, expect } from 'vitest'; +import { Dataset, Distribution } from '@lde/dataset'; +import { DataFactory } from 'n3'; +import { + datasetBindingsFor, + predefinedDatasetBindings, +} from '../../src/sparql/datasetBindings.js'; + +const { namedNode, literal } = DataFactory; + +const dataset = namedNode('http://example.org/dataset'); +const publisher = { + iri: new URL('http://example.org/publisher'), + name: { nl: 'Naam', en: 'Name' }, +}; + +function makeDataset( + overrides: Partial[0]> = {}, +) { + const distribution = Distribution.sparql( + new URL('http://example.org/sparql'), + ); + return new Dataset({ + iri: new URL('http://example.org/dataset'), + distributions: [distribution], + ...overrides, + }); +} + +describe('predefinedDatasetBindings', () => { + const datasetPublisher = namedNode('http://example.org/publisher'); + + it('binds only the dataset IRI when no optional properties are present', () => { + expect(predefinedDatasetBindings(makeDataset({ language: [] }))).toEqual([ + [{ dataset }], + [], + [], + ]); + }); + + it('groups datasetPublisher and datasetLicense with the dataset IRI', () => { + const groups = predefinedDatasetBindings( + makeDataset({ + publisher: { ...publisher, name: {} }, + license: new URL('http://example.org/license'), + }), + ); + + expect(groups).toEqual([ + [ + { + dataset, + datasetPublisher, + datasetLicense: namedNode('http://example.org/license'), + }, + ], + [], + [], + ]); + }); + + it('keeps languages and publisher names in independent groups', () => { + const groups = predefinedDatasetBindings( + makeDataset({ language: ['nl', 'en'], publisher }), + ); + + expect(groups).toEqual([ + [{ dataset, datasetPublisher }], + [{ datasetLanguage: literal('nl') }, { datasetLanguage: literal('en') }], + [ + { datasetPublisherName: literal('Naam', 'nl') }, + { datasetPublisherName: literal('Name', 'en') }, + ], + ]); + }); +}); + +describe('datasetBindingsFor', () => { + const type = namedNode('urn:Type'); + + it('adds custom rows as one more group', () => { + expect(datasetBindingsFor(makeDataset(), [{ type }])).toEqual([ + [{ dataset }], + [], + [], + [{ type }], + ]); + }); + + it('lets a custom row replace a predefined binding of the same name', () => { + const other = namedNode('urn:other'); + + expect( + datasetBindingsFor(makeDataset({ language: ['nl'] }), [ + { datasetLanguage: other }, + ]), + ).toEqual([[{ dataset }], [{}], [], [{ datasetLanguage: other }]]); + }); +}); diff --git a/packages/pipeline/test/sparql/reader.test.ts b/packages/pipeline/test/sparql/reader.test.ts index b5af8b51..5a854593 100644 --- a/packages/pipeline/test/sparql/reader.test.ts +++ b/packages/pipeline/test/sparql/reader.test.ts @@ -4,6 +4,11 @@ import { LineBufferTransform, readQueryFile, } from '../../src/sparql/index.js'; +import { + datasetBindingsFor, + predefinedDatasetBindings, +} from '../../src/sparql/datasetBindings.js'; +import type { ReadOptions } from '../../src/sparql/reader.js'; import { Dataset, Distribution } from '@lde/dataset'; import { startSparqlEndpoint, @@ -53,7 +58,7 @@ describe('SparqlConstructReader', () => { const reader = new SparqlConstructReader({ query: `CONSTRUCT { - ?dataset ?p ?o . + <${datasetIri}> ?p ?o . } WHERE { <${datasetIri}> ?p ?o . @@ -112,7 +117,7 @@ describe('SparqlConstructReader', () => { ); }); - it('substitutes ?dataset with dataset IRI', async () => { + it('binds ?dataset to the dataset IRI via a VALUES clause', async () => { const fetcher = new SparqlEndpointFetcher(); const querySpy = vi.spyOn(fetcher, 'fetchTriples'); @@ -131,15 +136,21 @@ describe('SparqlConstructReader', () => { distributions: [distribution], }); - await reader.read(dataset, distribution); + await reader.read(dataset, distribution, { + datasetBindings: predefinedDatasetBindings(dataset), + }); + expect(querySpy).toHaveBeenCalledWith( + expect.any(String), + expect.stringContaining('VALUES'), + ); expect(querySpy).toHaveBeenCalledWith( expect.any(String), expect.stringContaining(`<${datasetIri}>`), ); expect(querySpy).toHaveBeenCalledWith( expect.any(String), - expect.not.stringContaining('?dataset'), + expect.stringContaining('?dataset'), ); }); @@ -285,7 +296,7 @@ describe('SparqlConstructReader', () => { ); }); - it('does not inject a VALUES clause without bindings', async () => { + it('injects no VALUES clause without selector bindings', async () => { const fetcher = new SparqlEndpointFetcher(); const querySpy = vi.spyOn(fetcher, 'fetchTriples'); @@ -311,7 +322,7 @@ describe('SparqlConstructReader', () => { ); }); - it('does not inject a VALUES clause when bindings array is empty', async () => { + it('injects no VALUES clause when the selector bindings array is empty', async () => { const fetcher = new SparqlEndpointFetcher(); const querySpy = vi.spyOn(fetcher, 'fetchTriples'); @@ -338,6 +349,62 @@ describe('SparqlConstructReader', () => { }); }); + describe('datasetBindings', () => { + const datasetIri = 'http://foo.org/id/dataset/foo'; + const distribution = () => + Distribution.sparql( + new URL(`http://localhost:${port}/sparql`), + 'http://foo.org/id/graph/foo', + ); + const dataset = () => + new Dataset({ + iri: new URL(datasetIri), + distributions: [distribution()], + language: ['nl', 'en'], + }); + + async function read(query: string, options: ReadOptions) { + const reader = new SparqlConstructReader({ query }); + const result = await reader.read(dataset(), distribution(), options); + const triples: string[] = []; + for await (const quad of result) { + triples.push(`${quad.subject.value} ${quad.object.value}`); + } + return triples; + } + + it('joins a single custom row with a multi-valued binding, visible outside a sub-SELECT', async () => { + const triples = await read( + `CONSTRUCT { + ?s a ?type . + ?type ?datasetLanguage . + } + WHERE { + { SELECT ?s WHERE { ?s ?p ?o } } + }`, + { + bindings: [{ s: namedNode(datasetIri) }], + datasetBindings: datasetBindingsFor(dataset(), [ + { type: namedNode('urn:A') }, + ]), + }, + ); + + expect(new Set(triples)).toEqual( + new Set([`${datasetIri} urn:A`, 'urn:A nl', 'urn:A en']), + ); + }); + + it('binds nothing when no dataset bindings are passed', async () => { + const triples = await read( + 'CONSTRUCT { ?datasetPublisher ; } WHERE {}', + {}, + ); + + expect(triples).toEqual(['urn:1 urn:2']); + }); + }); + describe('retry', () => { it('retries on 504 and succeeds on second attempt', async () => { const fetcher = new SparqlEndpointFetcher(); @@ -488,7 +555,7 @@ describe('SparqlConstructReader', () => { const reader = new SparqlConstructReader({ query: `CONSTRUCT { - ?dataset ?p ?o . + <${datasetIri}> ?p ?o . } WHERE { <${datasetIri}> ?p ?o . diff --git a/packages/pipeline/test/sparql/selector.test.ts b/packages/pipeline/test/sparql/selector.test.ts index f1a3a12a..4723623f 100644 --- a/packages/pipeline/test/sparql/selector.test.ts +++ b/packages/pipeline/test/sparql/selector.test.ts @@ -753,4 +753,45 @@ describe('SparqlItemSelector', () => { expect(queries[2]).not.toContain('urn:graph:1'); }); }); + + describe('datasetBindings', () => { + it('injects dataset bindings on every page', async () => { + 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: + 'SELECT ?uri WHERE { ?uri ?dataset } ORDER BY ?uri', + fetcher: fetcher as never, + }); + + for await (const _row of selector.select(distribution, 2, { + datasetBindings: [ + [{ dataset: namedNode('http://example.com/dataset') }], + ], + })) { + // consume + } + + expect(queries).toHaveLength(2); + expect(queries[0]).toContain(''); + expect(queries[1]).toContain(''); + }); + + it('injects nothing when datasetBindings is omitted', async () => { + const { fetcher, queries } = pagedFetcher([[]]); + const selector = new SparqlItemSelector({ + query, + fetcher: fetcher as never, + }); + + await selectAll(selector, 10); + + expect(queries[0]).not.toContain('VALUES'); + }); + }); }); diff --git a/packages/pipeline/test/sparql/values.test.ts b/packages/pipeline/test/sparql/values.test.ts index e6bbc2e4..00c02f61 100644 --- a/packages/pipeline/test/sparql/values.test.ts +++ b/packages/pipeline/test/sparql/values.test.ts @@ -7,11 +7,17 @@ import type { PatternGroup, PatternUnion, QueryConstruct, + QuerySelect, PatternValues, } from '@traqula/rules-sparql-1-1'; -import { findSubSelect, injectValues } from '../../src/sparql/values.js'; - -const { namedNode } = DataFactory; +import type { DatasetBindings } from '../../src/sparql/datasetBindings.js'; +import { + findSubSelect, + injectValues, + injectDatasetBindings, +} from '../../src/sparql/values.js'; + +const { namedNode, literal } = DataFactory; const parser = new Parser(); const generator = new Generator(); @@ -320,4 +326,331 @@ describe('injectValues', () => { } }); }); + + it.each([ + ['plain', literal('nl'), { value: 'nl', langOrIri: undefined }], + [ + 'typed', + literal('1900', namedNode('http://www.w3.org/2001/XMLSchema#integer')), + { + value: '1900', + langOrIri: { + subType: 'namedNode', + value: 'http://www.w3.org/2001/XMLSchema#integer', + }, + }, + ], + [ + 'language-tagged', + literal('Naam', 'nl'), + { value: 'Naam', langOrIri: 'nl' }, + ], + ])('injects a %s literal binding', (_kind, term, expected) => { + const result = injectValues(baseQuery, [{ class: term }]); + + const values = result.where.patterns.find( + (p): p is PatternValues => p.subType === 'values', + ); + expect(values!.values[0]['class']).toMatchObject({ + type: 'term', + subType: 'literal', + ...expected, + }); + }); +}); + +describe('unsafe IRIs', () => { + const unsafe = namedNode('urn:x> } ?s ?p ?o . VALUES ?z { { + expect(() => injectValues(parseConstruct(query), [{ s: unsafe }])).toThrow( + 'unsafe characters', + ); + }); + + it('rejects an unsafe IRI in a dataset binding the query references', () => { + expect(() => + injectDatasetBindings(parseConstruct(query), [[{ o: unsafe }]]), + ).toThrow('unsafe characters'); + }); + + it('rejects an unsafe datatype IRI', () => { + expect(() => + injectValues(parseConstruct(query), [{ s: literal('1900', unsafe) }]), + ).toThrow('unsafe characters'); + }); + + it('ignores an unsafe IRI in a dataset binding the query does not reference', () => { + const parsed = parseConstruct(query); + + expect(injectDatasetBindings(parsed, [[{ unused: unsafe }]])).toBe(parsed); + }); +}); + +describe('injectDatasetBindings', () => { + const dataset = { dataset: namedNode('http://example.org/dataset') }; + const languages = [ + { datasetLanguage: literal('nl') }, + { datasetLanguage: literal('en') }, + ]; + const publisherNames = [ + { datasetPublisherName: literal('Naam', 'nl') }, + { datasetPublisherName: literal('Name', 'en') }, + ]; + const bindings = [[dataset], languages, publisherNames]; + + function inject(sparql: string): string { + return generator.generate( + injectDatasetBindings(parseConstruct(sparql), bindings), + ); + } + + it('leaves a query that references no binding unchanged', () => { + const query = parseConstruct('CONSTRUCT { ?s a ?c } WHERE { ?s a ?c }'); + + expect(injectDatasetBindings(query, bindings)).toBe(query); + }); + + it('binds a variable referenced only in the template', () => { + const generated = inject( + 'CONSTRUCT { ?s ?dataset } WHERE { ?s a ?c }', + ); + + expect(generated).toContain('VALUES'); + expect(generated).toContain(''); + expect(generated).not.toContain('datasetLanguage'); + }); + + it('binds a variable referenced only in the WHERE clause', () => { + const generated = inject( + 'CONSTRUCT { ?s a ?c } WHERE { ?s a ?c ; ?l FILTER(?l = ?datasetLanguage) }', + ); + + expect(generated).toContain('"nl"'); + expect(generated).not.toContain('UNION'); + expect(generated).not.toContain(''); + }); + + it('does not bind a variable that merely starts with a binding name', () => { + const query = parseConstruct( + 'CONSTRUCT { ?s ?datasetType } WHERE { ?s a ?datasetType }', + ); + + expect(injectDatasetBindings(query, bindings)).toBe(query); + }); + + it('leaves a referenced variable without a binding unbound rather than UNDEF', () => { + const result = injectDatasetBindings( + parseConstruct( + 'CONSTRUCT { ?dataset ?datasetLicense } WHERE { }', + ), + bindings, + ); + + expect(result.where.patterns[0]).toMatchObject({ + subType: 'values', + variables: [{ value: 'dataset' }], + }); + expect(generator.generate(result)).not.toContain('UNDEF'); + }); + + it('does not bind a variable that is out of scope inside a sub-SELECT', () => { + const query = parseConstruct( + 'CONSTRUCT { ?s a ?c } WHERE { { SELECT ?s ?c { ?s a ?c ; ?datasetLanguage } } }', + ); + + expect(injectDatasetBindings(query, bindings)).toBe(query); + }); + + it('binds a variable a sub-SELECT * exposes', () => { + const generated = inject( + 'CONSTRUCT { ?s a ?c } WHERE { { SELECT * { ?s a ?c ; ?datasetLanguage } } }', + ); + + expect(generated).toContain('"nl"'); + }); + + it('does not bind a variable a sub-SELECT only aggregates', () => { + const query = parseConstruct( + 'CONSTRUCT { ?s ?n } WHERE { { SELECT ?s (COUNT(?datasetLanguage) AS ?n) { ?s ?datasetLanguage } GROUP BY ?s } }', + ); + + expect(injectDatasetBindings(query, bindings)).toBe(query); + }); + + it('puts independent multi-valued groups in UNION branches, with single-valued ones outside', () => { + const result = injectDatasetBindings( + parseConstruct( + 'CONSTRUCT { ?dataset ?datasetLanguage ; ?datasetPublisherName } WHERE { }', + ), + bindings, + ); + + const [single, union] = result.where.patterns; + expect(single).toMatchObject({ + subType: 'values', + values: [{ dataset: { value: 'http://example.org/dataset' } }], + }); + expect(union.subType).toBe('union'); + const branches = (union as unknown as { patterns: PatternGroup[] }) + .patterns; + expect(branches.map((branch) => branch.patterns[0])).toMatchObject([ + { subType: 'values', values: [{}, {}] }, + { subType: 'values', values: [{}, {}] }, + ]); + expect(() => parser.parse(generator.generate(result))).not.toThrow(); + }); + + describe('multi-valued groups the query depends on', () => { + const types = [{ type: literal('A') }, { type: literal('B') }]; + + function patternsOf(sparql: string, groups: DatasetBindings = bindings) { + return injectDatasetBindings(parseConstruct(sparql), groups).where + .patterns; + } + + it('joins groups used in the WHERE clause', () => { + const generated = inject( + 'CONSTRUCT { ?s a ?c } WHERE { ?s ?datasetLanguage ; ?datasetPublisherName }', + ); + + expect(generated).not.toContain('UNION'); + expect(generated).toContain('"nl"'); + expect(generated).toContain('"Naam"@nl'); + }); + + it('joins a group used in a FILTER with one used only in the template', () => { + const generated = inject( + 'CONSTRUCT { ?s ?datasetPublisherName } WHERE { ?s ?l FILTER(?l = ?datasetLanguage) }', + ); + + expect(generated).not.toContain('UNION'); + expect(generated).toContain('"nl"'); + expect(generated).toContain('"Naam"@nl'); + }); + + it('counts a variable in a sub-SELECT projection as used in the WHERE clause', () => { + const generated = inject( + 'CONSTRUCT { ?s ?datasetPublisherName } WHERE { { SELECT ?s ?datasetLanguage { ?s ?datasetLanguage } } }', + ); + + expect(generated).not.toContain('UNION'); + expect(generated).toContain('"nl"'); + expect(generated).toContain('"Naam"@nl'); + }); + + it('joins groups a SELECT references, as it has no template', () => { + const select = parser.parse( + 'SELECT ?s WHERE { ?s ?datasetLanguage ; ?datasetPublisherName }', + ) as QuerySelect; + + const generated = generator.generate( + injectDatasetBindings(select, bindings), + ); + + expect(generated).not.toContain('UNION'); + expect(generated).toContain('"nl"'); + expect(generated).toContain('"Naam"@nl'); + }); + + it('keeps the other groups in UNION branches when one must be joined', () => { + const [language, union] = patternsOf( + 'CONSTRUCT { ?s ?datasetPublisherName . ?s ?type } WHERE { ?s ?datasetLanguage }', + [[dataset], languages, publisherNames, types], + ); + + expect(language).toMatchObject({ + subType: 'values', + variables: [{ value: 'datasetLanguage' }], + }); + expect(union.subType).toBe('union'); + }); + + it('joins only the groups that share a template triple', () => { + const [names, nameTypes, languageValues] = patternsOf( + 'CONSTRUCT { ?s ?datasetLanguage . ?datasetPublisherName ?type } WHERE { ?s a ?c }', + [[dataset], languages, publisherNames, types], + ); + + expect(names).toMatchObject({ + variables: [{ value: 'datasetPublisherName' }], + }); + expect(nameTypes).toMatchObject({ variables: [{ value: 'type' }] }); + expect(languageValues).toMatchObject({ + subType: 'values', + variables: [{ value: 'datasetLanguage' }], + }); + }); + }); + + it('uses plain VALUES when only one multi-valued group is referenced', () => { + const generated = inject( + 'CONSTRUCT { ?s ?datasetPublisherName } WHERE { ?s a ?c }', + ); + + expect(generated).not.toContain('UNION'); + expect(generated).toContain('"Naam"@nl'); + expect(generated).toContain('"Name"@en'); + }); + + it.each([ + [ + 'are equal for the referenced variables', + 'CONSTRUCT { ?s ?a } WHERE { ?s a ?c }', + [ + { a: literal('1'), b: literal('x') }, + { a: literal('1'), b: literal('y') }, + ], + ], + [ + 'bind the same terms in a different order', + 'CONSTRUCT { ?s ?a ; ?b } WHERE { ?s a ?c }', + [ + { a: literal('1'), b: literal('2') }, + { b: literal('2'), a: literal('1') }, + ], + ], + ])('collapses rows that %s', (_description, sparql, rows) => { + const result = injectDatasetBindings(parseConstruct(sparql), [rows]); + + expect((result.where.patterns[0] as PatternValues).values).toHaveLength(1); + }); + + it('keeps every variable of rows that bind different variables', () => { + const result = injectDatasetBindings( + parseConstruct( + 'CONSTRUCT { ?s ?a ; ?b } WHERE { ?s a ?c }', + ), + [[{ a: literal('1') }, { b: literal('2') }]], + ); + + expect( + (result.where.patterns[0] as PatternValues).variables.map( + (variable) => variable.value, + ), + ).toEqual(['a', 'b']); + expect(() => parser.parse(generator.generate(result))).not.toThrow(); + }); + + it('generates valid SPARQL when the WHERE clause is a bare sub-SELECT', () => { + const generated = inject( + 'CONSTRUCT { ?dataset ?count } WHERE { SELECT (COUNT(*) AS ?count) { ?s ?p ?o } }', + ); + + expect(() => parser.parse(generated)).not.toThrow(); + }); + + it('injects at the outer scope, not into a nested sub-SELECT', () => { + const query = parseConstruct( + 'CONSTRUCT { ?dataset ?partition } WHERE { { SELECT ?class { ?s a ?class } } BIND(IRI(CONCAT(STR(?dataset), STR(?class))) AS ?partition) }', + ); + + const result = injectDatasetBindings(query, bindings); + + expect(hasValues(result.where.patterns)).toBe(true); + expect( + hasValues(findSubSelect(result.where.patterns)!.where.patterns), + ).toBe(false); + }); }); diff --git a/packages/pipeline/test/stage.test.ts b/packages/pipeline/test/stage.test.ts index e80f5ac1..022d955b 100644 --- a/packages/pipeline/test/stage.test.ts +++ b/packages/pipeline/test/stage.test.ts @@ -7,6 +7,7 @@ import type { ItemSelector, QuadTransform, RunOptions, + SelectOptions, } from '../src/stage.js'; import type { Reader, ReadOptions } from '../src/sparql/reader.js'; import { NotSupported } from '../src/sparql/reader.js'; @@ -1338,4 +1339,59 @@ describe('Stage', () => { expect(reader.read.mock.calls[0][1]).toBe(distribution); }); }); + + describe('bindingsProvider', () => { + const datasetRow = { dataset: namedNode('http://example.org/dataset') }; + const customRow = { type: namedNode('http://example.org/CustomType') }; + const bindingsProvider = { bindings: async () => [customRow] }; + + it('hands the selector and readers the same dataset bindings, separate from item bindings', async () => { + const seenOptions: SelectOptions[] = []; + const item = { s: namedNode('http://example.org/s1') }; + const selector: ItemSelector = { + async *select(_distribution, _batchSize, options) { + seenOptions.push(options ?? {}); + yield item; + }, + }; + const reader = capturingExecutor([q1]); + const stage = new Stage({ + name: 'withProvider', + readers: reader, + itemSelector: selector, + bindingsProvider, + }); + + await stage.run(dataset, distribution, collectingWriter()); + + const options = reader.read.mock.calls[0][2] as ReadOptions; + expect(options.bindings).toEqual([item]); + expect(options.datasetBindings).toContainEqual([customRow]); + expect(seenOptions[0].datasetBindings).toBe(options.datasetBindings); + }); + + it('passes dataset bindings to the reader when there is no item selector', async () => { + const reader = capturingExecutor([q1]); + const stage = new Stage({ + name: 'withProvider', + readers: reader, + bindingsProvider, + }); + + await stage.run(dataset, distribution, collectingWriter()); + + const options = reader.read.mock.calls[0][2] as ReadOptions; + expect(options.datasetBindings).toContainEqual([customRow]); + }); + + it('passes the predefined bindings without a provider', async () => { + const reader = capturingExecutor([q1]); + const stage = new Stage({ name: 'withoutProvider', readers: reader }); + + await stage.run(dataset, distribution, collectingWriter()); + + const options = reader.read.mock.calls[0][2] as ReadOptions; + expect(options.datasetBindings).toContainEqual([datasetRow]); + }); + }); }); diff --git a/packages/pipeline/vite.config.ts b/packages/pipeline/vite.config.ts index 596338d3..41f3bccd 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: 98.98, - lines: 97.95, - branches: 93.34, - statements: 97.65, + functions: 99.1, + lines: 98.08, + branches: 93.94, + statements: 97.82, }, }, }, diff --git a/packages/search-pipeline/src/extraction.ts b/packages/search-pipeline/src/extraction.ts index d751b9df..aa9d75de 100644 --- a/packages/search-pipeline/src/extraction.ts +++ b/packages/search-pipeline/src/extraction.ts @@ -37,8 +37,9 @@ export interface ExtractionOptions { * The variable the roots bind to. Left free in the generated query for the * pipeline’s VALUES injection (`injectValues`), so it must match the stage’s * `rootVariable` – the variable the item selector projects. Defaults to - * `root`, matching `selectByClass`. Must not be `dataset` (the SPARQL reader - * substitutes `?dataset` with the dataset IRI). + * `root`, matching `selectByClass`. Must not be one of `@lde/pipeline`’s + * `DATASET_BINDING_VARIABLES` (`dataset`, `datasetPublisher`, …): the SPARQL + * reader binds those per dataset. * @default 'root' */ readonly subjectVariable?: string; diff --git a/packages/search-pipeline/src/search-stages.ts b/packages/search-pipeline/src/search-stages.ts index 4a8eb8c2..0a1d7b24 100644 --- a/packages/search-pipeline/src/search-stages.ts +++ b/packages/search-pipeline/src/search-stages.ts @@ -26,9 +26,11 @@ export interface SearchStageType { searchType: RootType; /** * 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`. + * the batch is complete for. Must **not** be one of + * `DATASET_BINDING_VARIABLES` (`dataset`, `datasetPublisher`, …): the item + * selector and SPARQL reader bind those per dataset, so a root bound to one + * would collide with that binding and never reach the projection. + * {@link selectByClass} defaults to `root`. */ rootVariable: string; /**