From 49219f68cdcfab6ded2a2b3389b048d2f428e0fe Mon Sep 17 00:00:00 2001 From: David de Boer Date: Fri, 16 Oct 2026 18:30:48 +0200 Subject: [PATCH 1/2] feat(sparql-qlever): report the QLever that built each index MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - Add an optional generic engine ({ name, version }) to ImportSuccessful - Read QLever’s git-hash from meta-data.json, for fresh and cached indexes - For a cached index, report the QLever that built it, not the one serving it --- docs/reference/sparql-importer.md | 5 +-- docs/reference/sparql-qlever.md | 2 ++ packages/sparql-importer/src/index.ts | 13 +++++++ packages/sparql-qlever/src/importer.ts | 36 +++++++++++++++----- packages/sparql-qlever/test/importer.test.ts | 22 ++++++++++-- packages/sparql-qlever/vite.config.ts | 6 ++-- 6 files changed, 68 insertions(+), 16 deletions(-) diff --git a/docs/reference/sparql-importer.md b/docs/reference/sparql-importer.md index e5fe2435..c3118471 100644 --- a/docs/reference/sparql-importer.md +++ b/docs/reference/sparql-importer.md @@ -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). diff --git a/docs/reference/sparql-qlever.md b/docs/reference/sparql-qlever.md index 8a8539ac..18a89765 100644 --- a/docs/reference/sparql-qlever.md +++ b/docs/reference/sparql-qlever.md @@ -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: diff --git a/packages/sparql-importer/src/index.ts b/packages/sparql-importer/src/index.ts index 1d34b425..ad564c95 100644 --- a/packages/sparql-importer/src/index.ts +++ b/packages/sparql-importer/src/index.ts @@ -36,9 +36,22 @@ export class ImportSuccessful { public readonly identifier?: string, public readonly tripleCount?: number, public readonly warnings: string[] = [], + /** The engine that built the imported data, when the importer can tell. + * For a cached index this is the engine that built it, which may differ + * from the one that serves it. */ + public readonly engine?: Engine, ) {} } +/** A store engine, identified as precisely as it identifies itself. */ +export interface Engine { + /** Human-readable name, e.g. `QLever`. */ + readonly name: string; + /** The engine’s own version identifier, such as a release number or a + * commit hash. */ + readonly version: string; +} + export class ImportFailed { constructor( public readonly distribution: Distribution, diff --git a/packages/sparql-qlever/src/importer.ts b/packages/sparql-qlever/src/importer.ts index 38b5ffcd..9e9d6be4 100644 --- a/packages/sparql-qlever/src/importer.ts +++ b/packages/sparql-qlever/src/importer.ts @@ -1,4 +1,5 @@ import { + type Engine, Importer as ImporterInterface, ImporterOptions, ImportFailed, @@ -102,14 +103,21 @@ export class Importer implements ImporterInterface { await this.options.downloader.download(distribution); if (await this.isIndexUpToDate(localFile)) { - const tripleCount = await this.readTripleCount(localFile); + const metadata = await this.readMetadata(localFile); + const tripleCount = this.parseTripleCount(metadata); if (tripleCount === 0) { return new ImportFailed( distribution, 'Index is cached but contains 0 triples', ); } - return new ImportSuccessful(distribution, undefined, tripleCount); + return new ImportSuccessful( + distribution, + undefined, + tripleCount, + [], + this.parseEngine(metadata), + ); } let indexFile = localFile; @@ -153,7 +161,13 @@ export class Importer implements ImporterInterface { await this.writeCacheInfo(localFile); - return new ImportSuccessful(distribution, undefined, tripleCount, warnings); + return new ImportSuccessful( + distribution, + undefined, + tripleCount, + warnings, + this.parseEngine(logs), + ); } private parseTripleCount(logs: string): number | undefined { @@ -164,6 +178,13 @@ export class Importer implements ImporterInterface { return match ? Number(match[1]) : undefined; } + /** The QLever that built the index, from the `git-hash` in its metadata: + * QLever has no releases, so the commit is its version. */ + private parseEngine(metadata: string): Engine | undefined { + const match = metadata.match(/"git-hash"\s*:\s*"([0-9a-f]+)"/); + return match ? { name: 'QLever', version: match[1] } : undefined; + } + private cacheInfoPath(dataFile: string): string { return join(dirname(dataFile), `${this.options.indexName}.cache-info.json`); } @@ -197,17 +218,16 @@ export class Importer implements ImporterInterface { return true; } - /** Read the triple count from QLever's metadata file. */ - private async readTripleCount(dataFile: string): Promise { + /** Read QLever's metadata file, or an empty string when there is none. */ + private async readMetadata(dataFile: string): Promise { try { const metadataPath = join( dirname(dataFile), `${this.options.indexName}.meta-data.json`, ); - const raw = await readFile(metadataPath, 'utf-8'); - return this.parseTripleCount(raw); + return await readFile(metadataPath, 'utf-8'); } catch { - return undefined; + return ''; } } diff --git a/packages/sparql-qlever/test/importer.test.ts b/packages/sparql-qlever/test/importer.test.ts index e935d1a5..b8c16183 100644 --- a/packages/sparql-qlever/test/importer.test.ts +++ b/packages/sparql-qlever/test/importer.test.ts @@ -39,7 +39,7 @@ function stubTaskRunner(tripleCount = 42): TaskRunner & { return command; }, async wait() { - return `{"num-triples":{"normal":${tripleCount}}}`; + return `{"git-hash":"a14e0a","num-triples":{"normal":${tripleCount}}}`; }, async stop() { return null; @@ -79,6 +79,10 @@ describe('Importer', () => { const result = await importer.import(distributions); expect(result).toBeInstanceOf(ImportSuccessful); expect((result as ImportSuccessful).tripleCount).toBe(1); + expect((result as ImportSuccessful).engine).toEqual({ + name: 'QLever', + version: expect.stringMatching(/^[0-9a-f]+$/), + }); }, 120_000); }); @@ -87,11 +91,14 @@ describe('Importer', () => { let dataFile: string; const indexName = 'test-index'; - /** Write the QLever metadata file that `readTripleCount` reads on cache hits. */ + /** Write the QLever metadata file that a cache hit reads. */ async function writeMetadata(tripleCount: number) { await writeFile( join(tempDir, `${indexName}.meta-data.json`), - JSON.stringify({ 'num-triples': { normal: tripleCount } }), + JSON.stringify({ + 'git-hash': '9352d0', + 'num-triples': { normal: tripleCount }, + }), ); } @@ -129,6 +136,10 @@ describe('Importer', () => { expect(result).toBeInstanceOf(ImportSuccessful); expect((result as ImportSuccessful).tripleCount).toBe(42); + expect((result as ImportSuccessful).engine).toEqual({ + name: 'QLever', + version: 'a14e0a', + }); expect(runner.commands.length).toBe(1); // Cache marker should exist with only the source file name. @@ -154,6 +165,11 @@ describe('Importer', () => { const result = await importer.import(makeDistributions()); expect(result).toBeInstanceOf(ImportSuccessful); expect((result as ImportSuccessful).tripleCount).toBe(42); + // The QLever that built the cached index, not the one that would index now. + expect((result as ImportSuccessful).engine).toEqual({ + name: 'QLever', + version: '9352d0', + }); expect(runner.commands.length).toBe(1); // Still only one index call. }); diff --git a/packages/sparql-qlever/vite.config.ts b/packages/sparql-qlever/vite.config.ts index a7c6c331..c5a2e6c5 100644 --- a/packages/sparql-qlever/vite.config.ts +++ b/packages/sparql-qlever/vite.config.ts @@ -13,10 +13,10 @@ export default mergeConfig( coverage: { thresholds: { autoUpdate: true, - lines: 96.25, + lines: 96.29, functions: 100, - branches: 88.49, - statements: 96.29, + branches: 88.69, + statements: 96.33, }, }, }, From 2dfcef8759d76066333cdea7ec79ca6122796d25 Mon Sep 17 00:00:00 2001 From: David de Boer Date: Fri, 16 Oct 2026 18:30:56 +0200 Subject: [PATCH 2/2] feat(pipeline)!: pass import details to reporters as one object MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - Replace ResolvedDistribution’s importedFrom, importDuration and tripleCount with imported - Pass that object to ProgressReporter.distributionSelected, now with the import engine - Print the engine in the console reporter: “Imported into QLever 9352d0 (…)” BREAKING CHANGE: ResolvedDistribution takes an ImportDetails object ({ from, duration, tripleCount?, engine? }) as its third argument instead of importedFrom, importDuration and tripleCount, and exposes it as imported. ProgressReporter.distributionSelected(dataset, distribution, imported?) receives the same object. --- docs/reference/pipeline.md | 44 ++++----- .../src/consoleReporter.ts | 21 +++-- .../test/consoleReporter.test.ts | 49 +++++++--- .../pipeline-console-reporter/vite.config.ts | 6 +- .../src/distribution/importResolver.ts | 19 ++-- packages/pipeline/src/distribution/index.ts | 1 + .../pipeline/src/distribution/resolver.ts | 19 +++- packages/pipeline/src/pipeline.ts | 27 +++--- packages/pipeline/src/progressReporter.ts | 7 +- .../test/distribution/importResolver.test.ts | 20 +++-- packages/pipeline/test/pipeline.test.ts | 90 +++++++++---------- packages/pipeline/vite.config.ts | 2 +- 12 files changed, 167 insertions(+), 138 deletions(-) diff --git a/docs/reference/pipeline.md b/docs/reference/pipeline.md index 2fed1cd3..335b562c 100644 --- a/docs/reference/pipeline.md +++ b/docs/reference/pipeline.md @@ -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 diff --git a/packages/pipeline-console-reporter/src/consoleReporter.ts b/packages/pipeline-console-reporter/src/consoleReporter.ts index 6ac1b3ce..b11e3a60 100644 --- a/packages/pipeline-console-reporter/src/consoleReporter.ts +++ b/packages/pipeline-console-reporter/src/consoleReporter.ts @@ -1,6 +1,7 @@ import type { Dataset, Distribution } from '@lde/dataset'; import type { DistributionAnalysisResult, + ImportDetails, ProgressReporter, TimeoutTransitionEvent, ValidationReport, @@ -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 = ''; diff --git a/packages/pipeline-console-reporter/test/consoleReporter.test.ts b/packages/pipeline-console-reporter/test/consoleReporter.test.ts index 3147138c..72ee5cff 100644 --- a/packages/pipeline-console-reporter/test/consoleReporter.test.ts +++ b/packages/pipeline-console-reporter/test/consoleReporter.test.ts @@ -210,12 +210,14 @@ 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(''); @@ -223,6 +225,29 @@ describe('ConsoleReporter', () => { 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(); @@ -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(''); diff --git a/packages/pipeline-console-reporter/vite.config.ts b/packages/pipeline-console-reporter/vite.config.ts index ce852ced..884a35c3 100644 --- a/packages/pipeline-console-reporter/vite.config.ts +++ b/packages/pipeline-console-reporter/vite.config.ts @@ -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, }, }, }, diff --git a/packages/pipeline/src/distribution/importResolver.ts b/packages/pipeline/src/distribution/importResolver.ts index 6d7a8a32..a6b84e18 100644 --- a/packages/pipeline/src/distribution/importResolver.ts +++ b/packages/pipeline/src/distribution/importResolver.ts @@ -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. * @@ -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) { diff --git a/packages/pipeline/src/distribution/index.ts b/packages/pipeline/src/distribution/index.ts index e452dece..80a0220f 100644 --- a/packages/pipeline/src/distribution/index.ts +++ b/packages/pipeline/src/distribution/index.ts @@ -19,6 +19,7 @@ export { ProbedDistributions, SparqlDistributionResolver, type DistributionResolver, + type ImportDetails, type ProbedSource, type ResolveCallbacks, type SparqlDistributionResolverOptions, diff --git a/packages/pipeline/src/distribution/resolver.ts b/packages/pipeline/src/distribution/resolver.ts index b340768d..b177c6d8 100644 --- a/packages/pipeline/src/distribution/resolver.ts +++ b/packages/pipeline/src/distribution/resolver.ts @@ -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, @@ -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, diff --git a/packages/pipeline/src/pipeline.ts b/packages/pipeline/src/pipeline.ts index f8618583..fff0363a 100644 --- a/packages/pipeline/src/pipeline.ts +++ b/packages/pipeline/src/pipeline.ts @@ -692,11 +692,11 @@ export class Pipeline { // the analysis stages is empirically incapable. Switch to the dataset’s // data dump and re-run all stages locally, discarding the // endpoint-sourced partial results. Only a live endpoint - // (`importedFrom === undefined`) can fall back – a run already on an + // (`imported === undefined`) can fall back – a run already on an // imported dump has nowhere further to go. if ( stageFailed && - resolved.importedFrom === undefined && + resolved.imported === undefined && this.distributionResolver.resolveFallback ) { // A failing fallback import must abort only this dataset, never the @@ -718,16 +718,13 @@ export class Pipeline { // run can skip an unchanged dump (the endpoint's fingerprint is // null, which would force a re-import every run). let fallbackFingerprint = fingerprint; - if (fallback.importedFrom) { + if (fallback.imported) { + const dump = fallback.imported.from; const dumpProbeResult = probed.probeResults.find( - (result) => - result.url === fallback.importedFrom!.accessUrl.toString(), + (result) => result.url === dump.accessUrl.toString(), ); if (dumpProbeResult) { - fallbackFingerprint = sourceFingerprint( - fallback.importedFrom, - dumpProbeResult, - ); + fallbackFingerprint = sourceFingerprint(dump, dumpProbeResult); } } this.reportSelectedDistribution( @@ -889,19 +886,17 @@ export class Pipeline { this.reporter?.distributionSelected?.( dataset, resolved.distribution, - resolved.importedFrom, - resolved.importDuration, - resolved.tripleCount, + resolved.imported, ); - if (resolved.importedFrom) { + if (resolved.imported) { this.reporter?.distributionValidated?.( - resolved.importedFrom, + resolved.imported.from, importOutcomeToVerdict( new ImportSuccessful( - resolved.importedFrom, + resolved.imported.from, undefined, - resolved.tripleCount, + resolved.imported.tripleCount, ), fingerprint, ), diff --git a/packages/pipeline/src/progressReporter.ts b/packages/pipeline/src/progressReporter.ts index 4ef78bca..cfc8c149 100644 --- a/packages/pipeline/src/progressReporter.ts +++ b/packages/pipeline/src/progressReporter.ts @@ -1,6 +1,7 @@ import type { Dataset, Distribution } from '@lde/dataset'; import type { ValidityVerdict } from '@lde/distribution-health'; import type { ValidationReport } from './validator.js'; +import type { ImportDetails } from './distribution/resolver.js'; import type { TimeoutTransitionEvent } from './sparql/timeoutPolicy.js'; export interface DistributionAnalysisResult { @@ -46,12 +47,12 @@ export interface ProgressReporter { distribution: Distribution, verdict: ValidityVerdict, ): void; + /** Called when a dataset’s source is resolved; `imported` is set when the + * source serves an imported data dump. */ distributionSelected?( dataset: Dataset, distribution: Distribution, - importedFrom?: Distribution, - importDuration?: number, - tripleCount?: number, + imported?: ImportDetails, ): void; stageStart?(stage: string): void; stageProgress?(update: { diff --git a/packages/pipeline/test/distribution/importResolver.test.ts b/packages/pipeline/test/distribution/importResolver.test.ts index a8b59df6..168105cc 100644 --- a/packages/pipeline/test/distribution/importResolver.test.ts +++ b/packages/pipeline/test/distribution/importResolver.test.ts @@ -138,6 +138,8 @@ describe('ImportResolver', () => { Distribution.sparql(new URL('http://localhost:7878/sparql')), 'test-graph', 42000, + [], + { name: 'QLever', version: '9352d0' }, ), ), }; @@ -160,10 +162,14 @@ describe('ImportResolver', () => { ); expect(resolved.probeResults).toHaveLength(1); expect(resolved.probeResults[0]).toBeInstanceOf(DataDumpProbeResult); - expect(resolved.tripleCount).toBe(42000); + expect(resolved.imported?.tripleCount).toBe(42000); + expect(resolved.imported?.engine).toEqual({ + name: 'QLever', + version: '9352d0', + }); }); - it('sets importedFrom on ResolvedDistribution when import succeeds', async () => { + it('sets imported on ResolvedDistribution when import succeeds', async () => { const dataset = makeDataset(); const inner = innerNoSparqlEndpoint(dataset, [dataDumpProbeResult]); @@ -185,10 +191,10 @@ describe('ImportResolver', () => { const result = await resolve(resolver, dataset); const resolved = result as ResolvedDistribution; - expect(resolved.importedFrom).toBe(importedDistribution); + expect(resolved.imported?.from).toBe(importedDistribution); }); - it('importedFrom is undefined when a SPARQL endpoint is used directly', async () => { + it('imported is undefined when a SPARQL endpoint is used directly', async () => { const dataset = makeDataset(); const distribution = Distribution.sparql(new URL(SPARQL_URL)); const inner = innerWithSparqlEndpoint(dataset, distribution, [ @@ -203,7 +209,7 @@ describe('ImportResolver', () => { const result = await resolve(resolver, dataset); expect(result).toBeInstanceOf(ResolvedDistribution); - expect((result as ResolvedDistribution).importedFrom).toBeUndefined(); + expect((result as ResolvedDistribution).imported).toBeUndefined(); }); it('returns NoDistributionAvailable with importFailed when import fails', async () => { @@ -414,8 +420,8 @@ describe('ImportResolver', () => { expect(resolved.distribution.accessUrl.toString()).toBe( 'http://localhost:7001/sparql', ); - expect(resolved.importedFrom).toBe(importedDistribution); - expect(resolved.tripleCount).toBe(99); + expect(resolved.imported?.from).toBe(importedDistribution); + expect(resolved.imported?.tripleCount).toBe(99); }); it('returns NoDistributionAvailable without importing when disabled', async () => { diff --git a/packages/pipeline/test/pipeline.test.ts b/packages/pipeline/test/pipeline.test.ts index 83ccdee4..c1ab8d7e 100644 --- a/packages/pipeline/test/pipeline.test.ts +++ b/packages/pipeline/test/pipeline.test.ts @@ -1271,19 +1271,17 @@ describe('Pipeline', () => { }); }); - it('distributionSelected reports importedFrom when import was used', async () => { + it('distributionSelected reports the import when one was used', async () => { const reporter = makeReporter(); const importedFromDistribution = new Distribution( new URL('http://example.org/data.nt'), 'application/n-triples', ); - const resolved = new ResolvedDistribution( - sparqlDistribution, - [], - importedFromDistribution, - 1000, - 42000, - ); + const resolved = new ResolvedDistribution(sparqlDistribution, [], { + from: importedFromDistribution, + duration: 1000, + tripleCount: 42000, + }); const pipeline = new Pipeline({ datasetSelector: makeDatasetSelector(dataset), @@ -1298,9 +1296,7 @@ describe('Pipeline', () => { expect(reporter.distributionSelected).toHaveBeenCalledWith( dataset, sparqlDistribution, - importedFromDistribution, - 1000, - 42000, + { from: importedFromDistribution, duration: 1000, tripleCount: 42000 }, ); }); @@ -1310,13 +1306,11 @@ describe('Pipeline', () => { new URL('http://example.org/data.nt'), 'application/n-triples', ); - const resolved = new ResolvedDistribution( - sparqlDistribution, - [], - importedFromDistribution, - 1000, - 42000, - ); + const resolved = new ResolvedDistribution(sparqlDistribution, [], { + from: importedFromDistribution, + duration: 1000, + tripleCount: 42000, + }); const resolver: DistributionResolver = { probe: vi.fn( @@ -1551,11 +1545,10 @@ describe('Pipeline', () => { ), resolveFallback: vi.fn( async () => - new ResolvedDistribution( - importedDistribution, - [], - dumpDistribution, - ), + new ResolvedDistribution(importedDistribution, [], { + from: dumpDistribution, + duration: 0, + }), ), }; } @@ -1664,7 +1657,10 @@ describe('Pipeline', () => { // The dataset resolves directly to an already-imported dump. resolver.resolve = vi.fn( async () => - new ResolvedDistribution(importedDistribution, [], dumpDistribution), + new ResolvedDistribution(importedDistribution, [], { + from: dumpDistribution, + duration: 0, + }), ); const pipeline = new Pipeline({ @@ -1696,11 +1692,10 @@ describe('Pipeline', () => { resolver.resolveFallback.mockImplementation( async (_probed, callbacks?: ResolveCallbacks) => { callbacks?.onImportStart?.(); - return new ResolvedDistribution( - importedDistribution, - [], - dumpDistribution, - ); + return new ResolvedDistribution(importedDistribution, [], { + from: dumpDistribution, + duration: 0, + }); }, ); const reporter = makeReporter(); @@ -1805,9 +1800,7 @@ describe('Pipeline', () => { expect(reporter.distributionSelected).toHaveBeenLastCalledWith( dataset, importedDistribution, - dumpDistribution, - undefined, - undefined, + { from: dumpDistribution, duration: 0 }, ); expect(reporter.distributionValidated).toHaveBeenCalledWith( dumpDistribution, @@ -1852,7 +1845,7 @@ describe('Pipeline', () => { it('re-runs against a non-imported fallback source without a deep verdict', async () => { // A custom resolver may fall back to another live source (e.g. a secondary - // endpoint) rather than an imported dump: no importedFrom, so no deep + // endpoint) rather than an imported dump: no import, so no deep // validity verdict and the endpoint's fingerprint is kept. const secondaryEndpoint = Distribution.sparql( new URL('http://secondary.example.org/sparql'), @@ -1892,8 +1885,6 @@ describe('Pipeline', () => { dataset, secondaryEndpoint, undefined, - undefined, - undefined, ); // No imported distribution → no deep validity verdict for the fallback. expect(reporter.distributionValidated).not.toHaveBeenCalled(); @@ -2547,9 +2538,11 @@ describe('Pipeline', () => { | NoDistributionAvailable = new ResolvedDistribution( sparqlDistribution, [dumpProbe], - dumpDistribution, - 1, - 100, + { + from: dumpDistribution, + duration: 1, + tripleCount: 100, + }, ), ): DistributionResolver & { probe: ReturnType; @@ -2808,13 +2801,11 @@ describe('Pipeline', () => { it('surfaces an empty deep verdict when an import yields no triples', async () => { const resolver = makeDumpResolver( - new ResolvedDistribution( - sparqlDistribution, - [dumpProbe], - dumpDistribution, - 1, - 0, - ), + new ResolvedDistribution(sparqlDistribution, [dumpProbe], { + from: dumpDistribution, + duration: 1, + tripleCount: 0, + }), ); const reporter = makeReporter(); @@ -3236,14 +3227,13 @@ describe('beforeDatasetWrite hook', () => { resolve: vi.fn(async () => new ResolvedDistribution(endpoint, [])), resolveFallback: vi.fn( async () => - new ResolvedDistribution( - dump, - [], - new Distribution( + new ResolvedDistribution(dump, [], { + from: new Distribution( new URL('http://example.org/dump.ttl'), 'application/n-triples', ), - ), + duration: 0, + }), ), }; // Writes q1 on the endpoint then hard-fails; writes q2 on the dump re-run. diff --git a/packages/pipeline/vite.config.ts b/packages/pipeline/vite.config.ts index 9fbf1188..956c0453 100644 --- a/packages/pipeline/vite.config.ts +++ b/packages/pipeline/vite.config.ts @@ -14,7 +14,7 @@ export default mergeConfig( functions: 97.48, lines: 97.18, branches: 92.05, - statements: 96.74, + statements: 96.73, }, }, },