Composite Stages #12

New issue

Open

opened 2026-07-23 15:15:06 +00:00 by erik · 0 comments

erik commented

2026-07-23 15:15:06 +00:00

Owner

The performance optimization should be a later phase: once composition semantics are stable, add optional transform-chain fusion for safe linear transform-only composites.

Key Changes

Extend external template files to support composite entries alongside existing single-stage templates.

Add a composite template shape:

  stdin-jsonl:
    stages:
      read:
        module: stdin
      lines:
        module: parse-lines
        inputs: [read]
      json:
        module: parse-json
        inputs: [lines]
    output: json

Allow pipeline usage as one public stage:

  pipeline:
    ingest:
      template: local.stdin-jsonl
    write:
      module: stdout
      inputs: [ingest]

During config expansion:
- Generate hidden internal stage names from the public stage name, e.g. ingest::read, ingest::lines, ingest::json.
- Rewrite downstream references to ingest so they consume the declared composite output.
- Rewrite internal inputs to point at hidden names.
- Support $inputs inside composite internal stage inputs for transform composites that receive the public stage’s external inputs.
- Reject unknown internal inputs, missing output, invalid $inputs usage, duplicate generated stage names, and composites that would shadow reserved built-in channels.
Keep runtime behavior unchanged in v1: each internal stage is still a normal source/transform/sink with normal stats, error routing, UI visibility, and backpressure.

Public Interface

Existing single-stage templates continue to work unchanged.
Composite templates are only loaded from declared external template files.
For composite override config, use internal stage names:
```
  pipeline:
    ingest:
      template: local.stdin-jsonl
      config:
        lines:
          max-size: 131072
```
Each override deep-merges into that internal stage’s config; unknown override keys are startup errors.

Performance Follow-Up

Add optional fusion later for linear transform-only composites.
Fusion should only apply when every internal stage is a Transform using default start semantics or explicitly marked fusible.
Do not fuse sources, sinks, observability subscribers, or stages that depend on timer/shutdown behavior until the trait model exposes those capabilities safely.
Fused composites should preserve per-stage error names or add clear composite context so debugging does not get worse.

Test Plan

Unit-test composite expansion:
- source composite expands to hidden internal stages.
- downstream inputs: [ingest] resolves to composite output.
- $inputs wires public external inputs into the first internal transform.
- invalid output/internal input/override names fail at startup.
- existing single-stage templates are unaffected.
Add an e2e pipeline proving the motivating case:
- composite stdin-jsonl handles multiple JSON lines from stdin.
- direct stdin -> parse-json remains invalid or emits parse errors for chunked/multiple-line payloads.
Run:
- just check
- targeted e2e for the new composite pipeline
- just validate before merging

Assumptions

v1 prioritizes footgun prevention over runtime performance.
Composite definitions live in external template files, not in the standard plugin set.
The first implementation should be config expansion, not a new StagePlugin::Composite runtime type.

While stage templates help with boiler-plate of common stages, it doesn't help with common stage composition. One example is reading json from stdin. The stdin plugin emits chunks of text, so it needs a parse-lines stage between it and parse-json. Another example is that organizations will have a common set of transforms to apply to most all pipelines: normalize, sensitive data scrub, enrichment. Abstracting that into a single composite stage can make the pipeline easier to reason about but more importantly help avoid potential footguns. The performance optimization should be a later phase: once composition semantics are stable, add optional transform-chain fusion for safe linear transform-only composites. ## Key Changes - Extend external template files to support composite entries alongside existing single-stage templates. - Add a composite template shape: ```yaml stdin-jsonl: stages: read: module: stdin lines: module: parse-lines inputs: [read] json: module: parse-json inputs: [lines] output: json ``` - Allow pipeline usage as one public stage: ```yaml pipeline: ingest: template: local.stdin-jsonl write: module: stdout inputs: [ingest] ``` - During config expansion: - Generate hidden internal stage names from the public stage name, e.g. ingest::read, ingest::lines, ingest::json. - Rewrite downstream references to ingest so they consume the declared composite output. - Rewrite internal inputs to point at hidden names. - Support $inputs inside composite internal stage inputs for transform composites that receive the public stage’s external inputs. - Reject unknown internal inputs, missing output, invalid $inputs usage, duplicate generated stage names, and composites that would shadow reserved built-in channels. - Keep runtime behavior unchanged in v1: each internal stage is still a normal source/transform/sink with normal stats, error routing, UI visibility, and backpressure. ## Public Interface - Existing single-stage templates continue to work unchanged. - Composite templates are only loaded from declared external template files. - For composite override config, use internal stage names: ```yaml pipeline: ingest: template: local.stdin-jsonl config: lines: max-size: 131072 ``` Each override deep-merges into that internal stage’s config; unknown override keys are startup errors. ## Performance Follow-Up - Add optional fusion later for linear transform-only composites. - Fusion should only apply when every internal stage is a Transform using default start semantics or explicitly marked fusible. - Do not fuse sources, sinks, observability subscribers, or stages that depend on timer/shutdown behavior until the trait model exposes those capabilities safely. - Fused composites should preserve per-stage error names or add clear composite context so debugging does not get worse. ## Test Plan - Unit-test composite expansion: - source composite expands to hidden internal stages. - downstream inputs: [ingest] resolves to composite output. - $inputs wires public external inputs into the first internal transform. - invalid output/internal input/override names fail at startup. - existing single-stage templates are unaffected. - Add an e2e pipeline proving the motivating case: - composite stdin-jsonl handles multiple JSON lines from stdin. - direct stdin -> parse-json remains invalid or emits parse errors for chunked/multiple-line payloads. - Run: - just check - targeted e2e for the new composite pipeline - just validate before merging ## Assumptions - v1 prioritizes footgun prevention over runtime performance. - Composite definitions live in external template files, not in the standard plugin set. - The first implementation should be config expansion, not a new StagePlugin::Composite runtime type.

erik self-assigned this

2026-07-23 15:15:06 +00:00