feat(pipeline)!: support dataset and user-provided bindings

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
 

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 <graph>` 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.

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<Stage>, 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,
+  );
+});

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<VariableBindings[]>;
+}

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';