Skip to content

feat(search-pipeline): extract root types from the dataset registry #698

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
ddeboer merged 2 commits into from
Aug 7, 2026
+869 −68
Merged

feat(search-pipeline): extract root types from the dataset registry #698

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
111a293
feat(search-pipeline): extract root types from the dataset registry
ddeboer Aug 5, 2026
0f1661c
fix(pipeline): reject unsafe graph IRIs and sourced chained stages
ddeboer Aug 7, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
71 changes: 53 additions & 18 deletions docs/reference/search-indexer.md
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -64,24 +64,25 @@ images.

## Configuration

| Variable | Default | Meaning |
| -------------------- | --------------------------- | ----------------------------------------------------------------------------------------------------------------------------------------------------- |
| `SCHEMA_MODULE` | `/config/search-schema.mjs` | Path of the mounted schema-declaration module |
| `REGISTRY_ENDPOINT` | **required** | SPARQL endpoint of the DCAT dataset registry |
| `DATASETS` | all registry datasets | Dataset IRIs to index (whitespace- or comma-separated) |
| `DATASET_CRITERIA` | all registry datasets | Search criteria as a JSON object, in [`@lde/dataset-registry-client`](./dataset-registry-client)’s criteria format (mutually exclusive w/ `DATASETS`) |
| `TYPESENSE_HOST` | **required** | Typesense host |
| `TYPESENSE_PORT` | `8108` | Typesense port |
| `TYPESENSE_PROTOCOL` | `http` | `http` or `https` |
| `TYPESENSE_API_KEY` | **required** | An admin key: the indexer creates, writes and swaps collections |
| `REBUILD_MODE` | `in-place` | `in-place` (update the live collection) or `blue-green` (swap on commit) |
| `COLLECTION_PREFIX` | none | Prefix for every derived collection name (configure the read side to match) |
| `PROVENANCE_FILE` | none | JSON file remembering per-dataset processing, to skip unchanged datasets |
| `PIPELINE_VERSION` | none | Version keying the skip decisions; required with `PROVENANCE_FILE` |
| `QLEVER_IMAGE` | none | Enables the QLever import path (see below), e.g. `adfreiburg/qlever:latest` |
| `IMPORT_STRATEGY` | `sparql` | `sparql`, `sparqlWithImportFallback` or `import`; requires `QLEVER_IMAGE` |
| `DATA_DIR` | `/data` | Directory for downloaded dumps and QLever index caches |
| `QLEVER_NETWORK` | none | Docker network the spawned QLever joins; set when the indexer itself runs containerized (see below) |
| Variable | Default | Meaning |
| --------------------- | --------------------------- | ----------------------------------------------------------------------------------------------------------------------------------------------------- |
| `SCHEMA_MODULE` | `/config/search-schema.mjs` | Path of the mounted schema-declaration module |
| `REGISTRY_ENDPOINT` | **required** | SPARQL endpoint of the DCAT dataset registry |
| `REGISTRY_ROOT_TYPES` | none | Root types extracted from the registry instead of the dataset’s distribution (see below), by schema type name |
| `DATASETS` | all registry datasets | Dataset IRIs to index (whitespace- or comma-separated) |
| `DATASET_CRITERIA` | all registry datasets | Search criteria as a JSON object, in [`@lde/dataset-registry-client`](./dataset-registry-client)’s criteria format (mutually exclusive w/ `DATASETS`) |
| `TYPESENSE_HOST` | **required** | Typesense host |
| `TYPESENSE_PORT` | `8108` | Typesense port |
| `TYPESENSE_PROTOCOL` | `http` | `http` or `https` |
| `TYPESENSE_API_KEY` | **required** | An admin key: the indexer creates, writes and swaps collections |
| `REBUILD_MODE` | `in-place` | `in-place` (update the live collection) or `blue-green` (swap on commit) |
| `COLLECTION_PREFIX` | none | Prefix for every derived collection name (configure the read side to match) |
| `PROVENANCE_FILE` | none | JSON file remembering per-dataset processing, to skip unchanged datasets |
| `PIPELINE_VERSION` | none | Version keying the skip decisions; required with `PROVENANCE_FILE` |
| `QLEVER_IMAGE` | none | Enables the QLever import path (see below), e.g. `adfreiburg/qlever:latest` |
| `IMPORT_STRATEGY` | `sparql` | `sparql`, `sparqlWithImportFallback` or `import`; requires `QLEVER_IMAGE` |
| `DATA_DIR` | `/data` | Directory for downloaded dumps and QLever index caches |
| `QLEVER_NETWORK` | none | Docker network the spawned QLever joins; set when the indexer itself runs containerized (see below) |

A misconfigured boot reports **all** problems in one error, not one per crash
loop. `PROVENANCE_FILE` must sit on a durable volume, and cannot be combined
Expand All @@ -92,6 +93,40 @@ that ownership, so a named volume mounted there just works, while a bind
mount must be `chown`ed on the host. A run fails at start when the file is
not writable, instead of silently never skipping.

## Registry-sourced root types

Every root type is extracted from the dataset’s **distribution** by default –
its live SPARQL endpoint, or a dump imported into QLever. That holds while a
type is described by the data the dataset publishes, and it breaks for the
dataset itself: a dataset’s description is governed by a different application
profile from the objects it contains, and it lives in the register. Nothing
obliges a publisher to describe its own dataset inside its dump, so a `Dataset`
type extracted from the distribution yields documents for the few publishers
that self-describe and nothing for the rest.

`REGISTRY_ROOT_TYPES` names the root types to extract from `REGISTRY_ENDPOINT`
instead:

```sh
--env REGISTRY_ROOT_TYPES="Dataset Publisher"
```

Same CONSTRUCT generator, same framing, same projection, same writers – only
the source differs. Each such stage is **scoped to the graph the dataset in
hand names**: a register holds every registration, so an unscoped stage would
re-index the whole catalogue once per dataset processed. Scoped, one pass sees
exactly one registration, and a `selectByClass` finds that dataset’s own roots
inside it – its `dcat:Dataset` node, and the `foaf:Organization` its
`dcterms:publisher` points at. This presumes the register names each
registration’s graph after the dataset IRI, which is how a DCAT register that
crawls per registration stores it.

Routing is deployment topology, so it is configuration and not part of the
schema: a `SearchType` is defined by its `class`, never by where its triples
come from, and the same declaration serves a deployment that sources it
differently. A name that the mounted schema does not declare fails the boot,
rather than shipping an empty collection.

## The QLever import path

By default the indexer serves only datasets that publish a live SPARQL
Expand Down
70 changes: 70 additions & 0 deletions docs/reference/search-pipeline.md
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -171,6 +171,8 @@ const pipeline = new Pipeline<TypedSearchDocument>({
},
],
}),
// A stage reads the dataset’s own distribution unless `sourceFor` says
// otherwise – see “Registry-sourced root types” below.
// Each type gets its own collection – an independent blue/green rebuild –
// named by the adapter from the type itself (`Dataset` → `datasets`,
// `Organization` → `organizations`), so no naming map is passed here and the
Expand Down Expand Up @@ -224,6 +226,74 @@ blank-node subjects (`FILTER(!isBlank(?root))`): a blank node has no stable
document key, so it can never become a search document – a custom selector
should exclude them too.

## Registry-sourced root types

A stage reads the dataset’s own distribution. That holds while a root type is
described by the data the dataset publishes, and it breaks for the dataset
itself: a dataset’s description is governed by a different application profile
from the objects it contains, and it lives in the **dataset registry**.
Registering a dataset is submitting a description, so the register covers every
dataset a pipeline can select – while nothing obliges a publisher to describe
its own dataset inside its dump.

`registrySource` points a stage at the register instead, scoped to the graph the
dataset in hand names:

```typescript
import {
registrySource,
searchStages,
selectByClass,
} from '@lde/search-pipeline';

searchStages({
schema,
types: [
{
searchType: dataset,
rootVariable: 'root',
itemSelector: selectByClass(dataset),
sourceFor: registrySource(new URL('https://registry.example.org/sparql')),
},
// …the object-grain types keep reading the dataset’s distribution.
],
});
```

`searchIndexerPipeline` exposes the same routing as `registryTypes`, by type
name:

```typescript
searchIndexerPipeline({
schema,
datasets,
writerFor,
registryTypes: {
endpoint: new URL('https://registry.example.org/sparql'),
names: ['Dataset', 'Publisher'],
},
});
```

Same CONSTRUCT generator, same framing, same projection, same writers – only
the source differs. Two properties make it work:

- **The scoping is what keeps the stage per-dataset.** A register holds every
registration, so an unscoped stage would index the whole catalogue once per
dataset processed. `registrySource` scopes selection _and_ extraction to the
dataset’s graph, so one pass sees exactly one registration. It presumes the
register names each registration’s graph after the dataset IRI, which is how
a DCAT register that crawls per registration stores it.
- **The hops come from the declared paths.** No CBD rule and no hardcoded
predicate list: a `Publisher` root type with `label` at
`<http://xmlns.com/foaf/0.1/name>` resolves through the same mechanism as any
other type, because a registration describes its publisher inside its own
graph.

Routing is deployment topology, deliberately kept out of the `SearchType`: a
type is defined by its `class`, not by where its triples come from, so the same
declaration serves a deployment that sources it differently.

## Per-stage tuning

Each `SearchStageType` entry carries three knobs beyond its selector and
Expand Down
22 changes: 19 additions & 3 deletions packages/pipeline/src/sparql/graph.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -1,16 +1,32 @@
import { AstFactory, type QueryConstruct } from '@traqula/rules-sparql-1-1';
import { assertSafeIri } from '@lde/dataset';
import {
AstFactory,
type QueryConstruct,
type QuerySelect,
} from '@traqula/rules-sparql-1-1';

const F = new AstFactory();

/**
* Set the default graph (FROM clause) on a parsed CONSTRUCT query.
* Set the default graph (FROM clause) on a parsed CONSTRUCT or SELECT query.
*
* Mutates the query in place, replacing any existing FROM clause.
*
* Both query forms scope the same way, and a stage reading a
* {@link Distribution} with a `namedGraph` must scope **both**: were only the
* reader to honour it, an item selector would pick its roots from the whole
* endpoint and the reader would then find nothing for most of them.
*
* The graph IRI is {@link assertSafeIri}-checked before it becomes an `<…>`
* reference: a `namedGraph` is a plain string, typically carried in from
* third-party registry data, and one containing an angle bracket or whitespace
* would break out of the IRI reference and rewrite the query around it.
*/
export function withDefaultGraph(
query: QueryConstruct,
query: QueryConstruct | QuerySelect,
graphIri: string,
): void {
assertSafeIri(graphIri);
query.datasets = F.datasetClauses(
[{ clauseType: 'default', value: F.termNamed(F.gen(), graphIri) }],
F.gen(),
Expand Down
23 changes: 17 additions & 6 deletions packages/pipeline/src/sparql/selector.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -9,6 +9,7 @@ import {
type TermVariable,
} from '@traqula/rules-sparql-1-1';
import type { ItemSelector, SelectOptions } from '../stage.js';
import { withDefaultGraph } from './graph.js';
import type { VariableBindings } from './reader.js';
import {
ConstantTimeoutPolicy,
Expand All @@ -34,15 +35,15 @@ export interface SparqlItemSelectorOptions {
* SELECT query projecting at least one named variable.
*
* A `LIMIT` clause in the query overrides the stage's `batchSize` as the
* page size — use this when the SPARQL endpoint enforces a result limit.
* page size – use this when the SPARQL endpoint enforces a result limit.
* It does **not** cap the total number of bindings the selector yields;
* pagination continues with `OFFSET` until the source is exhausted. Use
* {@link maxResults} to cap the total.
*/
query: string;
/**
* Maximum number of bindings the selector yields across all pages.
* Use this for sampling — “give me at most N items, don’t walk the full
* Use this for sampling – “give me at most N items, don’t walk the full
* source”. Independent of {@link query}’s `LIMIT`, which controls page
* size. Pagination stops as soon as `maxResults` bindings have been
* yielded. Must not be negative.
Expand All @@ -64,8 +65,12 @@ export interface SparqlItemSelectorOptions {
* pagination early. Dropped rows are still fetched; to skip them at the
* endpoint, filter in the query itself (e.g. `FILTER(isIRI(?s))`).
*
* The endpoint URL comes from the {@link Distribution} passed to {@link select}.
* Pagination is an internal detail — consumers iterate binding rows directly.
* The endpoint URL comes from the {@link Distribution} passed to {@link select},
* as does the graph to scope to: a distribution declaring a `namedGraph` selects
* within that graph (a `FROM` clause), the same scoping `SparqlConstructReader`
* applies to the extraction. The two must scope alike, or the roots come from a
* wider source than the quads read for them.
* Pagination is an internal detail – consumers iterate binding rows directly.
*
* The page size (results per SPARQL request) is determined by, in order:
* 1. A `LIMIT` clause in the selector query (for endpoints with hard result limits)
Expand Down Expand Up @@ -129,6 +134,12 @@ export class SparqlItemSelector implements ItemSelector {
);
}
const endpoint = distribution.accessUrl!;
// Per call, so concurrent selections over different datasets cannot see
// each other's graph scope or page window.
const query = structuredClone(this.parsed);
if (distribution.namedGraph) {
withDefaultGraph(query, distribution.namedGraph);
}
const policy = options?.timeout ?? defaultTimeoutPolicy;
let offset = 0;
let totalFetched = 0;
Expand All @@ -148,12 +159,12 @@ export class SparqlItemSelector implements ItemSelector {
offset === 0 || totalFetched > totalYielded
? basePageSize
: Math.min(basePageSize, remaining);
this.parsed.solutionModifiers.limitOffset = F.solutionModifierLimitOffset(
query.solutionModifiers.limitOffset = F.solutionModifierLimitOffset(
effectivePageSize,
offset,
F.gen(),
);
const paginatedQuery = generator.generate(this.parsed);
const paginatedQuery = generator.generate(query);

const stream = await this.fetchBindingsWithPolicy(
endpoint,
Expand Down
Loading