feat(pipeline-void)!: iterate property partitions per property

docs/reference/pipeline-void.md

@@ -16,14 +16,15 @@ Returns all VoID stages in their recommended execution order. The ordering is op
 
 Accepts an optional `VoidStagesOptions` object:
 
-| Option           | Default | Description                                                                                                     |
-| ---------------- | ------- | --------------------------------------------------------------------------------------------------------------- |
-| `batchSize`      | 10      | Maximum class bindings per reader call (per-class stages only)                                                  |
-| `maxConcurrency` | 10      | Maximum concurrent in-flight reader batches (per-class stages only)                                             |
-| `perClass`       | –       | Override per-class iteration for all five per-class stages                                                      |
-| `uriSpaces`      | –       | When provided, includes the object URI space stage                                                              |
-| `vocabularies`   | –       | Additional vocabulary namespace URIs to detect beyond the built-in defaults                                     |
-| `transforms`     | –       | Transforms to attach to bundled stages, keyed by `VOID_STAGE_NAMES` (see [Stage transforms](#stage-transforms)) |
+| Option           | Default | Description                                                                                                       |
+| ---------------- | ------- | ----------------------------------------------------------------------------------------------------------------- |
+| `batchSize`      | 10      | Maximum item bindings per reader call (per-class stages and the per-property vocabularies stage)                  |
+| `maxConcurrency` | 10      | Maximum concurrent in-flight reader batches (per-class stages and the per-property vocabularies stage)            |
+| `perClass`       | –       | Override per-class iteration for all five per-class stages                                                        |
+| `perProperty`    | `true`  | Iterate the vocabularies (property-partition) stage per property, bounding its memory by batch instead of dataset |
+| `uriSpaces`      | –       | When provided, includes the object URI space stage                                                                |
+| `vocabularies`   | –       | Additional vocabulary namespace URIs to detect beyond the built-in defaults                                       |
+| `transforms`     | –       | Transforms to attach to bundled stages, keyed by `VOID_STAGE_NAMES` (see [Stage transforms](#stage-transforms))   |
 
 Per-request timeouts are configured at the `Pipeline` level via `PipelineOptions.timeout`, not per VoID stage.
 
@@ -77,10 +78,10 @@ The class selector that drives per-class iteration honours the distribution’s
 
 #### Domain-specific stages:
 
-| Factory                  | Description                                                                                                                                                                                                                                                                                          |
-| ------------------------ | ---------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------- |
-| `detectVocabularies()`   | [`entity-properties.rq`](https://github.com/ldelements/lde/blob/main/packages/pipeline-void/queries/entity-properties.rq) – Entity properties with automatic `void:vocabulary` detection. Accepts `DetectVocabulariesOptions` with an optional `vocabularies` array to extend the built-in defaults. |
-| `uriSpaces(uriSpaceMap)` | [`object-uri-space.rq`](https://github.com/ldelements/lde/blob/main/packages/pipeline-void/queries/object-uri-space.rq) – Object URI namespace linksets, aggregated against a provided URI space map                                                                                                 |
+| Factory                  | Description                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                            |
+| ------------------------ | ------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------ |
+| `detectVocabularies()`   | [`entity-properties.rq`](https://github.com/ldelements/lde/blob/main/packages/pipeline-void/queries/entity-properties.rq) – Entity properties with automatic `void:vocabulary` detection. Accepts `DetectVocabulariesOptions`: an optional `vocabularies` array to extend the built-in defaults, plus `perProperty` (default `true`), `batchSize` and `maxConcurrency`. By default the query iterates per property – the single global `GROUP BY` over every property materializes the dataset’s full scan and exhausts the query-memory budget on large datasets (measured on a 608M-triple dataset); `perProperty: false` restores the single query. |
+| `uriSpaces(uriSpaceMap)` | [`object-uri-space.rq`](https://github.com/ldelements/lde/blob/main/packages/pipeline-void/queries/object-uri-space.rq) – Object URI namespace linksets, aggregated against a provided URI space map                                                                                                                                                                                                                                                                                                                                                                                                                                                   |
 
 ## Namespace normalization
 
@@ -105,11 +106,11 @@ Use `namespacePartitionMergePlugin(aliases)` for namespaces other than schema.or
 
 ## Stage transforms
 
-A VoID stage decorates its reader’s output with a `QuadTransform<ReaderContext>` attached as data (see [@lde/pipeline](./pipeline)’s extension model and [ADR 2](../decisions/0002-unify-pipeline-extension-on-quad-transforms)). It runs once per reader call and may fire its own SPARQL queries against the `distribution` in scope – so write it to accept being called more than once: a global stage calls it once over the complete output, a per-class stage with batching enabled once per batch (one class at `batchSize: 1`).
+A VoID stage decorates its reader’s output with a `QuadTransform<ReaderContext>` attached as data (see [@lde/pipeline](./pipeline)’s extension model and [ADR 2](../decisions/0002-unify-pipeline-extension-on-quad-transforms)). It runs once per reader call and may fire its own SPARQL queries against the `distribution` in scope – so write it to accept being called more than once: a global stage calls it once over the complete output, an iterating stage once per batch (one class – or, on the vocabularies stage, one property – at `batchSize: 1`).
 
 Two transform factories are built in:
 
-- `withVocabularies(vocabularies?)` – passes through all quads and appends `void:vocabulary` triples for detected vocabulary namespace prefixes in `void:property` quads. The built-in defaults are exported as `defaultVocabularies` (sourced from `@zazuko/prefixes`); `detectVocabularies()` attaches it to the `entity-properties.rq` stage.
+- `withVocabularies(vocabularies?)` – passes through all quads and appends `void:vocabulary` triples for detected vocabulary namespace prefixes in `void:property` quads. The built-in defaults are exported as `defaultVocabularies` (sourced from `@zazuko/prefixes`); `detectVocabularies()` attaches it to the `entity-properties.rq` stage. The transform remembers which vocabularies it already emitted per distribution instance, so per-property batches yield each `void:vocabulary` quad once per stage run – reuse one transform instance within a run, and expect a fresh distribution (e.g. a fallback re-run) to start clean.
 - `withUriSpaces(uriSpaceMap)` – consumes `void:Linkset` quads, matches each `void:objectsTarget` against the configured URI space prefixes using `startsWith`, and aggregates triple counts per matched space. Emits `void:objectsTarget` pointing to the target dataset IRI (taken from the metadata quad subjects), not the raw prefix; unmatched linksets are discarded. `uriSpaces(uriSpaceMap)` attaches it to the `object-uri-space.rq` stage.
 
 ### Attaching your own transform

packages/pipeline-void/src/stage.ts

@@ -57,8 +57,7 @@ export type VoidStageName =
 
 /** A transform, or transforms, decorating a VoID stage's reader output. */
 export type VoidStageTransform =
-  | QuadTransform<ReaderContext>
-  | QuadTransform<ReaderContext>[];
+  QuadTransform<ReaderContext> | QuadTransform<ReaderContext>[];
 
 /**
  * Options for configuring VoID stage execution.
@@ -82,11 +81,13 @@ export interface VoidStageOptions {
 /**
  * Options for per-class VoID stages that iterate over classes.
  *
- * `batchSize` and `maxConcurrency` control how class bindings are batched
- * and processed concurrently – they have no effect on global (non-per-class) stages.
+ * `batchSize` and `maxConcurrency` control how item bindings are batched and
+ * processed concurrently. They apply to every iterating stage – the five
+ * per-class stages and (via {@link VoidStagesOptions}) the per-property
+ * vocabularies stage – and have no effect on global one-query stages.
  */
 export interface PerClassVoidStageOptions extends VoidStageOptions {
-  /** Maximum number of class bindings per reader call. @default 10 */
+  /** Maximum number of item bindings per reader call. @default 10 */
   batchSize?: number;
   /** Maximum concurrent in-flight reader batches. @default 10 */
   maxConcurrency?: number;
@@ -109,6 +110,12 @@ export interface VoidStagesOptions extends Omit<
   uriSpaces?: ReadonlyMap<string, readonly Quad[]>;
   /** Additional vocabulary namespace URIs to detect beyond the built-in defaults. */
   vocabularies?: readonly string[];
+  /**
+   * When true, iterate the vocabularies (property-partition) stage per
+   * property (see {@link DetectVocabulariesOptions.perProperty}).
+   * @default true
+   */
+  perProperty?: boolean;
   /**
    * Transforms to attach to bundled stages, keyed by {@link VOID_STAGE_NAMES}.
    *
@@ -125,6 +132,7 @@ async function createVoidStage(
   options?: {
     transform?: VoidStageTransform;
     perClass?: boolean;
+    perProperty?: boolean;
     batchSize?: number;
     maxConcurrency?: number;
     expectsOutput?: boolean;
@@ -141,7 +149,11 @@ async function createVoidStage(
   return new Stage({
     name: filename,
     readers: reader,
-    itemSelector: options?.perClass ? classSelector() : undefined,
+    itemSelector: options?.perClass
+      ? classSelector()
+      : options?.perProperty
+        ? propertySelector()
+        : undefined,
     batchSize: options?.batchSize,
     maxConcurrency: options?.maxConcurrency,
     expectsOutput: options?.expectsOutput,
@@ -156,25 +168,35 @@ function asTransforms(
   return Array.isArray(transform) ? [...transform] : [transform];
 }
 
-function classSelector(): ItemSelector {
+/**
+ * Build an {@link ItemSelector} that pages through the distinct bindings of
+ * one variable.
+ *
+ * The `ORDER BY` is load-bearing: the inner {@link SparqlItemSelector} pages
+ * with `LIMIT`/`OFFSET` (the query’s `LIMIT` is the page size), and SPARQL
+ * guarantees no result order without `ORDER BY` – unordered pages can skip or
+ * repeat items on endpoints with unstable ordering.
+ */
+function distinctItemSelector(
+  variable: string,
+  pattern: (subjectFilter: string) => string,
+): ItemSelector {
   return {
     // Forward `options` so the Pipeline’s per-dataset TimeoutPolicy
     // reaches the inner SparqlItemSelector – without this the adaptive
-    // budget is silently bypassed for class selection.
+    // budget is silently bypassed for item selection.
     select: (distribution, batchSize, options) => {
       const subjectFilter = distribution.subjectFilter ?? '';
       let fromClause = '';
       if (distribution.namedGraph) {
         assertSafeIri(distribution.namedGraph);
         fromClause = `FROM <${distribution.namedGraph}>`;
       }
-      // Exclude blank-node classes at the endpoint: the selector would drop
-      // them client-side anyway (no stable identity), but only after fetching
-      // them.
       const selectorQuery = [
-        'SELECT DISTINCT ?class',
+        `SELECT DISTINCT ?${variable}`,
         fromClause,
-        `WHERE { ${subjectFilter} ?s a ?class . FILTER(!isBlank(?class)) }`,
+        `WHERE { ${pattern(subjectFilter)} }`,
+        `ORDER BY ?${variable}`,
         'LIMIT 1000',
       ].join('\n');
 
@@ -185,6 +207,37 @@ function classSelector(): ItemSelector {
   };
 }
 
+function classSelector(): ItemSelector {
+  // Exclude blank-node classes at the endpoint: the selector would drop
+  // them client-side anyway (no stable identity), but only after fetching
+  // them.
+  return distinctItemSelector(
+    'class',
+    (subjectFilter) =>
+      `${subjectFilter} ?s a ?class . FILTER(!isBlank(?class))`,
+  );
+}
+
+/**
+ * Select the distinct properties of a dataset, so the property-partition
+ * query can iterate in bounded batches instead of aggregating every property
+ * at once.
+ *
+ * The chunking exists for memory, not style: the unchunked
+ * `entity-properties.rq` computes two `COUNT(DISTINCT …)` aggregates grouped
+ * over every property in one query, which materializes the dataset’s full
+ * scan – measured to exhaust a 16 GB query budget on a 608M-triple dataset,
+ * so large datasets got no property partitions (and, downstream, no
+ * `void:vocabulary`) at all. With a property batch injected as `VALUES`, the
+ * working set is bounded by one batch’s triples regardless of dataset size.
+ */
+function propertySelector(): ItemSelector {
+  return distinctItemSelector(
+    'p',
+    (subjectFilter) => `${subjectFilter} ?s ?p ?o .`,
+  );
+}
+
 // Global stages
 
 export function subjectUriSpaces(options?: VoidStageOptions): Promise<Stage> {
@@ -309,20 +362,33 @@ export function uriSpaces(
   });
 }
 
-export interface DetectVocabulariesOptions extends VoidStageOptions {
+export interface DetectVocabulariesOptions extends Omit<
+  PerClassVoidStageOptions,
+  'perClass'
+> {
   /** Additional vocabulary namespace URIs to detect beyond the built-in defaults. */
   vocabularies?: readonly string[];
+  /**
+   * When true, iterate the property-partition query per property using a
+   * property selector, so its memory use is bounded by the batch instead of
+   * the whole dataset (see {@link propertySelector}). @default true
+   */
+  perProperty?: boolean;
 }
 
 export function detectVocabularies(
   options?: DetectVocabulariesOptions,
 ): Promise<Stage> {
-  const { vocabularies, transform } = options ?? {};
+  const { vocabularies, transform, batchSize, maxConcurrency, perProperty } =
+    options ?? {};
   const allVocabularies = vocabularies
     ? [...defaultVocabularies, ...vocabularies]
     : undefined;
   return createVoidStage(VOID_STAGE_NAMES.vocabularies, {
     transform: [withVocabularies(allVocabularies), ...asTransforms(transform)],
+    perProperty: perProperty ?? true,
+    batchSize,
+    maxConcurrency,
   });
 }
 
@@ -341,6 +407,10 @@ export async function voidStages(
     uriSpaces: uriSpaceMap,
     vocabularies,
     transforms,
+    // Destructured out of stageOptions so it cannot leak into the global
+    // stages’ createVoidStage options – it applies to the vocabularies stage
+    // only.
+    perProperty,
     ...stageOptions
   } = options ?? {};
 
@@ -378,6 +448,7 @@ export async function voidStages(
     detectVocabularies({
       ...withTransform(VOID_STAGE_NAMES.vocabularies),
       vocabularies,
+      perProperty,
     }),
     subjectUriSpaces(withTransform(VOID_STAGE_NAMES.subjectUriSpace)),
     ...(uriSpaceMap

packages/pipeline-void/src/vocabularyTransform.ts

@@ -1,4 +1,5 @@
 import type { ReaderContext, QuadTransform } from '@lde/pipeline';
+import type { Distribution } from '@lde/dataset';
 import type { Quad } from '@rdfjs/types';
 import prefixes from '@zazuko/prefixes';
 import { DataFactory } from 'n3';
@@ -22,18 +23,37 @@ export const defaultVocabularies: readonly string[] = [
  * Attach it to the `entity-properties.rq` stage's reader – directly via
  * {@link detectVocabularies} or through the `transforms` map of
  * {@link voidStages}.
+ *
+ * With per-property iteration the transform sees one batch at a time, so it
+ * remembers which vocabularies it already emitted per {@link Distribution}
+ * and yields each `void:vocabulary` quad only once per stage run. The memory
+ * is keyed weakly on the distribution instance – one per run – so a fallback
+ * re-run of the same dataset (a fresh distribution) starts clean.
  */
 export function withVocabularies(
   vocabularies: readonly string[] = defaultVocabularies,
 ): QuadTransform<ReaderContext> {
-  return (quads, { dataset }) =>
-    appendVocabularies(quads, dataset.iri.toString(), vocabularies);
+  const emittedPerRun = new WeakMap<Distribution, Set<string>>();
+  return (quads, { dataset, distribution }) => {
+    let emitted = emittedPerRun.get(distribution);
+    if (!emitted) {
+      emitted = new Set();
+      emittedPerRun.set(distribution, emitted);
+    }
+    return appendVocabularies(
+      quads,
+      dataset.iri.toString(),
+      vocabularies,
+      emitted,
+    );
+  };
 }
 
 async function* appendVocabularies(
   quads: AsyncIterable<Quad>,
   datasetIri: string,
   vocabularies: readonly string[],
+  emitted: Set<string>,
 ): AsyncIterable<Quad> {
   const detectedVocabularies = new Set<string>();
 
@@ -53,6 +73,10 @@ async function* appendVocabularies(
 
   const datasetNode = namedNode(datasetIri);
   for (const vocabUri of detectedVocabularies) {
+    if (emitted.has(vocabUri)) {
+      continue;
+    }
+    emitted.add(vocabUri);
     yield quad(datasetNode, _void.vocabulary, namedNode(vocabUri));
   }
 }