Skip to content

Report which QLever built each index #921

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
Oct 16, 2026
+235 −154
Merged

Report which QLever built each index #921

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
49219f6
feat(sparql-qlever): report the QLever that built each index
ddeboer Oct 16, 2026
2dfcef8
feat(pipeline)!: pass import details to reporters as one object
ddeboer Oct 16, 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
44 changes: 22 additions & 22 deletions docs/reference/pipeline.md
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -419,28 +419,28 @@ Each reporter receives every event, in array order; a reporter that does not imp

The full event set, in rough lifecycle order:

| Event | Fired |
| ------------------------------------------------------------------------------------------- | ----------------------------------------------------------------------------------------------------------------------------------------- |
| `pipelineStart(name)` | Once, when the run starts. |
| `datasetsSelected(count, duration)` | After the dataset selector returns, with the selection size and duration (ms). |
| `selectionEmpty(reason)` | When the run is abandoned because the selection was empty – no commit, destination unchanged (see [Empty selections](#empty-selections)). |
| `datasetStart(dataset)` | When a dataset’s processing begins; datasets are processed one at a time, so it precedes all of that dataset’s later events. |
| `distributionProbed(result)` | Each time a single distribution probe completes, with a `DistributionAnalysisResult`. |
| `importStarted()` | When a data-dump import begins. |
| `importFailed(distribution, error)` | When importing a distribution fails. |
| `distributionValidated(distribution, verdict)` | With the RDF-validity verdict for each distribution the pipeline attempted – even when the dataset is otherwise skipped. |
| `distributionSelected(dataset, distribution, importedFrom?, importDuration?, tripleCount?)` | When a dataset’s source is resolved; the optional arguments are set when it was imported from a dump. |
| `stageStart(stage)` | When a stage begins. |
| `stageProgress({ itemsProcessed, quadsGenerated, memoryUsageBytes, heapUsedBytes })` | After each batch a stage processes. |
| `stageComplete(stage, { itemsProcessed, quadsGenerated, duration })` | When a stage finishes successfully. |
| `stageSkipped(stage, reason)` | When a stage is [skipped](#skipped-stages) (`NotSupported` or an empty selector). |
| `stageFailed(stage, error)` | When a stage hard-fails; the run continues (see [Failure isolation](#failure-isolation)). |
| `datasetValidated(dataset, report)` | Once per (dataset, validator) pair after all stages, even when nothing was validated – read `quadsValidated` to distinguish ‘not tested’. |
| `datasetSkipped(dataset, reason)` | When a dataset is skipped: probe/resolve failure, no distribution available, or unchanged since the last run. |
| `datasetComplete(dataset, { memoryUsageBytes, heapUsedBytes })` | When a dataset’s processing ends. |
| `pipelineComplete({ duration, memoryUsageBytes, heapUsedBytes })` | Once, when the run ends. |
| `timeoutTightened(event)` | When a [timeout policy](#timeout-policies) tightens an endpoint’s budget after consecutive timeouts. |
| `timeoutRelaxed(event)` | When a timeout policy relaxes a previously-tightened endpoint back to the default budget. |
| Event | Fired |
| ------------------------------------------------------------------------------------ | ----------------------------------------------------------------------------------------------------------------------------------------- |
| `pipelineStart(name)` | Once, when the run starts. |
| `datasetsSelected(count, duration)` | After the dataset selector returns, with the selection size and duration (ms). |
| `selectionEmpty(reason)` | When the run is abandoned because the selection was empty – no commit, destination unchanged (see [Empty selections](#empty-selections)). |
| `datasetStart(dataset)` | When a dataset’s processing begins; datasets are processed one at a time, so it precedes all of that dataset’s later events. |
| `distributionProbed(result)` | Each time a single distribution probe completes, with a `DistributionAnalysisResult`. |
| `importStarted()` | When a data-dump import begins. |
| `importFailed(distribution, error)` | When importing a distribution fails. |
| `distributionValidated(distribution, verdict)` | With the RDF-validity verdict for each distribution the pipeline attempted – even when the dataset is otherwise skipped. |
| `distributionSelected(dataset, distribution, imported?)` | When a dataset’s source is resolved; `imported` (`{ from, duration, tripleCount?, engine? }`) is set when it was imported from a dump. |
| `stageStart(stage)` | When a stage begins. |
| `stageProgress({ itemsProcessed, quadsGenerated, memoryUsageBytes, heapUsedBytes })` | After each batch a stage processes. |
| `stageComplete(stage, { itemsProcessed, quadsGenerated, duration })` | When a stage finishes successfully. |
| `stageSkipped(stage, reason)` | When a stage is [skipped](#skipped-stages) (`NotSupported` or an empty selector). |
| `stageFailed(stage, error)` | When a stage hard-fails; the run continues (see [Failure isolation](#failure-isolation)). |
| `datasetValidated(dataset, report)` | Once per (dataset, validator) pair after all stages, even when nothing was validated – read `quadsValidated` to distinguish ‘not tested’. |
| `datasetSkipped(dataset, reason)` | When a dataset is skipped: probe/resolve failure, no distribution available, or unchanged since the last run. |
| `datasetComplete(dataset, { memoryUsageBytes, heapUsedBytes })` | When a dataset’s processing ends. |
| `pipelineComplete({ duration, memoryUsageBytes, heapUsedBytes })` | Once, when the run ends. |
| `timeoutTightened(event)` | When a [timeout policy](#timeout-policies) tightens an endpoint’s budget after consecutive timeouts. |
| `timeoutRelaxed(event)` | When a timeout policy relaxes a previously-tightened endpoint back to the default budget. |

### Provenance store

Expand Down
5 changes: 3 additions & 2 deletions docs/reference/sparql-importer.md
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -21,8 +21,9 @@ components program against. It ships no importer of its own.
supports, downloads it, and indexes it. It resolves to one of three results:

- `ImportSuccessful` – with the imported `distribution`, an optional `identifier`
(e.g. the index name), an optional `tripleCount`, and any `warnings` collected
along the way;
(e.g. the index name), an optional `tripleCount`, any `warnings` collected
along the way, and an optional `engine` (`{ name, version }`) naming the store
engine that built the imported data, when the importer can tell;
- `ImportFailed` – with the `distribution` that failed and the `error` message;
- `NotSupported` – when none of the distributions use a supported format
(`distribution` is then optional).
Expand Down
2 changes: 2 additions & 0 deletions docs/reference/sparql-qlever.md
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -95,6 +95,8 @@ For native formats, the actual `qlever-index` format flag is resolved by priorit

An import that indexes **0 triples** – fresh or from cache – is treated as `ImportFailed`, not success.

A successful import reports the QLever that built the index as `ImportSuccessful.engine`, e.g. `{ name: 'QLever', version: '9352d0' }`, with the `git-hash` from the index’s `meta-data.json` as the version. For a cached index this is the QLever that built it, which may be older than the one serving it.

## Direct construction

`createQlever()` is a convenience; construct `Importer` and `Server` directly to supply your own task runner:
Expand Down
21 changes: 10 additions & 11 deletions packages/pipeline-console-reporter/src/consoleReporter.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -1,6 +1,7 @@
import type { Dataset, Distribution } from '@lde/dataset';
import type {
DistributionAnalysisResult,
ImportDetails,
ProgressReporter,
TimeoutTransitionEvent,
ValidationReport,
Expand Down Expand Up @@ -143,20 +144,18 @@ export class ConsoleReporter implements ProgressReporter {
distributionSelected(
_dataset: Dataset,
distribution: Distribution,
importedFrom?: Distribution,
importDuration?: number,
tripleCount?: number,
imported?: ImportDetails,
): void {
if (importedFrom) {
if (imported) {
const engine = imported.engine
? ` into ${imported.engine.name} ${imported.engine.version}`
: '';
const count =
tripleCount !== undefined
? `${compactNumber.format(tripleCount)} triples, `
: '';
const duration =
importDuration !== undefined
? ` in ${chalk.bold(prettyMilliseconds(importDuration))}`
imported.tripleCount !== undefined
? `${compactNumber.format(imported.tripleCount)} triples, `
: '';
const text = `Imported ${importedFrom.accessUrl.toString()} (${count}to ${distribution.accessUrl.toString()})${duration}`;
const duration = ` in ${chalk.bold(prettyMilliseconds(imported.duration))}`;
const text = `Imported ${imported.from.accessUrl.toString()}${engine} (${count}to ${distribution.accessUrl.toString()})${duration}`;
if (this.activeSpinner) {
this.activeSpinner.text = text;
this.activeSpinner.suffixText = '';
Expand Down
49 changes: 38 additions & 11 deletions packages/pipeline-console-reporter/test/consoleReporter.test.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -210,19 +210,44 @@ describe('ConsoleReporter', () => {
reporter.distributionSelected(
makeDataset(),
Distribution.sparql(new URL('http://localhost:7001/sparql')),
new Distribution(
new URL('http://example.org/data.nt'),
'application/n-triples',
),
5000,
4800000,
{
from: new Distribution(
new URL('http://example.org/data.nt'),
'application/n-triples',
),
duration: 5000,
tripleCount: 4800000,
},
);

const output = spy.mock.calls.map((c) => String(c[0])).join('');
expect(output).toContain('4.8M triples');
expect(output).toContain('to http://localhost:7001/sparql');
});

it('names the engine that built the import when present', () => {
const reporter = new ConsoleReporter();
const spy = vi.spyOn(process.stderr, 'write').mockReturnValue(true);

reporter.distributionSelected(
makeDataset(),
Distribution.sparql(new URL('http://localhost:7001/sparql')),
{
from: new Distribution(
new URL('http://example.org/data.nt'),
'application/n-triples',
),
duration: 5000,
engine: { name: 'QLever', version: '9352d0' },
},
);

const output = spy.mock.calls.map((c) => String(c[0])).join('');
expect(output).toContain(
'Imported http://example.org/data.nt into QLever 9352d0 (to http://localhost:7001/sparql)',
);
});

it('prints "(selected)" without cursor escapes on non-TTY', () => {
// Tests run in non-TTY mode, so this exercises the non-TTY path.
const reporter = new ConsoleReporter();
Expand Down Expand Up @@ -251,11 +276,13 @@ describe('ConsoleReporter', () => {
reporter.distributionSelected(
makeDataset(),
Distribution.sparql(new URL('http://localhost:7001/sparql')),
new Distribution(
new URL('http://example.org/data.nt'),
'application/n-triples',
),
5000,
{
from: new Distribution(
new URL('http://example.org/data.nt'),
'application/n-triples',
),
duration: 5000,
},
);

const output = spy.mock.calls.map((c) => String(c[0])).join('');
Expand Down
6 changes: 3 additions & 3 deletions packages/pipeline-console-reporter/vite.config.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -11,9 +11,9 @@ export default mergeConfig(
thresholds: {
autoUpdate: true,
functions: 76,
lines: 70.64,
branches: 48.21,
statements: 70.9,
lines: 70.9,
branches: 50,
statements: 71.17,
},
},
},
Expand Down
19 changes: 9 additions & 10 deletions packages/pipeline/src/distribution/importResolver.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -23,14 +23,14 @@ export interface ImportResolverOptions {
* Controls how a dataset's distribution is selected, ordered by how eagerly a
* data dump is imported:
*
* - `'sparql'` (default) — use a dataset's own SPARQL endpoint when one is
* - `'sparql'` (default) – use a dataset's own SPARQL endpoint when one is
* available; import a data dump only when no endpoint responds.
* - `'sparqlWithImportFallback'` — like `'sparql'`, but additionally fall
* - `'sparqlWithImportFallback'` – like `'sparql'`, but additionally fall
* back to importing the data dump when the endpoint passes probing yet fails
* to serve an analysis stage at runtime. The pipeline then re-runs all
* stages locally against the import (see
* {@link ImportResolver.resolveFallback}).
* - `'import'` — always import a data dump into a local SPARQL server,
* - `'import'` – always import a data dump into a local SPARQL server,
* even when the dataset advertises a working SPARQL endpoint. Useful when
* the remote endpoint is too slow or unreliable.
*
Expand Down Expand Up @@ -215,13 +215,12 @@ export class ImportResolver implements DistributionResolver {
);
distribution.subjectFilter = importResult.distribution.subjectFilter;

return new ResolvedDistribution(
distribution,
probeResults,
importResult.distribution,
Date.now() - importStart,
importResult.tripleCount,
);
return new ResolvedDistribution(distribution, probeResults, {
from: importResult.distribution,
duration: Date.now() - importStart,
tripleCount: importResult.tripleCount,
engine: importResult.engine,
});
}

if (importResult instanceof ImportFailed) {
Expand Down
1 change: 1 addition & 0 deletions packages/pipeline/src/distribution/index.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -19,6 +19,7 @@ export {
ProbedDistributions,
SparqlDistributionResolver,
type DistributionResolver,
type ImportDetails,
type ProbedSource,
type ResolveCallbacks,
type SparqlDistributionResolverOptions,
Expand Down
19 changes: 15 additions & 4 deletions packages/pipeline/src/distribution/resolver.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -1,5 +1,5 @@
import { Dataset, Distribution } from '@lde/dataset';
import type { ImportFailed } from '@lde/sparql-importer';
import type { Engine, ImportFailed } from '@lde/sparql-importer';
import {
probe,
SparqlProbeResult,
Expand All @@ -10,12 +10,23 @@ export class ResolvedDistribution {
constructor(
readonly distribution: Distribution,
readonly probeResults: ProbeResultType[],
readonly importedFrom?: Distribution,
readonly importDuration?: number,
readonly tripleCount?: number,
/** Set when {@link distribution} serves an imported data dump. */
readonly imported?: ImportDetails,
) {}
}

/** How a data dump was imported into the store a dataset is read from. */
export interface ImportDetails {
/** The data dump that was imported. */
readonly from: Distribution;
/** Milliseconds the import took. */
readonly duration: number;
/** Triples in the imported data, when the importer can tell. */
readonly tripleCount?: number;
/** The engine that built the imported data, when the importer can tell. */
readonly engine?: Engine;
}

export class NoDistributionAvailable {
constructor(
readonly dataset: Dataset,
Expand Down
Loading
Loading