feat: first-class data assets, materializations, lineage, and freshness
- Dominant language
- Go
- Stars
- 4k
- Forks
- 337
- Avg merge
- 19h 26m
- Merged PRs (30d)
- 133
Description
## Is your feature request related to a problem?
Dagu currently models workflows, steps, runs, local build materializations, and run artifacts very well. Data pipelines, however, usually produce durable **data assets** whose identity outlives one DAG run: a database table, a DuckLake table, an object-store prefix, a Parquet dataset, an API snapshot, or a search index.
Without a semantic asset model, operators cannot answer basic questions without building a separate catalog:
- What is the latest successful version of this dataset?
- Which run and source versions produced it?
- Is it stale?
- Which downstream datasets depend on it?
- Did a run succeed but fail to publish its dataset?
- Which datasets should be refreshed after an upstream asset changes?
Build workflows provide local file caching, while #2636 proposes stable publication of named artifacts. Both are useful, but neither represents a durable dataset, its materialization history, or lineage across DAGs.
## Describe the solution you'd like
Add first-class data assets and materialization events. An illustrative definition could look like:
```yaml
assets:
inputs:
- key: source/companies-house/company-data
uri: s3://raw/companies-house/company-data/
outputs:
- key: canonical/companies
uri: ducklake://catalog/main/companies
freshness:
warn_after: 24h
fail_after: 48h
steps:
- id: build_companies
uses: duckdb@v1
with:
query: transform.sql
inputs:
- asset: source/companies-house/company-data
outputs:
- asset: canonical/companies
metadata:
row_count: ${BUILD_COMPANIES_OUTPUT.row_count}
max_source_date: ${BUILD_COMPANIES_OUTPUT.max_source_date}
```
The exact syntax is only illustrative. The important capabilities are:
1. **Stable asset identity**
- A globally unique asset key independent of DAG and step names.
- Optional type, URI, owner, description, tags, and custom metadata.
- Support both Dagu-produced assets and external/source assets.
2. **Declared inputs and outputs**
- DAG- or step-level asset dependencies.
- Cross-DAG lineage without requiring one DAG to invoke another.
- Validation for duplicate producers and invalid references.
3. **Materialization records**
- Record a materialization only after the producing step succeeds and publication is complete.
- Store run ID, step ID, attempt, trigger, timestamp, partition key if present, code/spec version, and upstream materialization versions.
- Allow structured metadata such as schema, row count, byte count, source watermark, checksum/fingerprint, snapshot ID, and user-defined fields.
- Distinguish successful runs from successful materializations.
4. **Version and lineage queries**
- API/CLI/UI access to latest successful materialization and history.
- Upstream/downstream graph across DAG boundaries.
- Links from an asset version to the producing run and from a run to its emitted assets.
5. **Freshness policies**
- Warn/fail thresholds based on the last successful materialization or a source-provided logical timestamp.
- Fresh/stale/unknown status in the UI and API.
- Prometheus/OpenTelemetry signals and notification hooks.
6. **Trigger integration**
- A future event-trigger implementation such as #2551 should be able to emit `asset.materialized` or `asset.changed` events.
- Downstream DAGs should optionally trigger from an asset event without being coupled through `dag.run`.
This should remain lightweight. A file-backed implementation is sufficient for an MVP; row-level lineage and a full data catalog are not required.
## Describe alternatives you've considered
- Encode asset names and versions in filenames or object-store paths.
- Store materialization metadata manually with Dagu state actions.
- Publish named artifacts through #2636.
- Run an external catalog such as OpenMetadata or DataHub.
- Couple pipelines by invoking downstream DAGs directly.
These approaches work, but each workflow must recreate identity, lineage, freshness, and publication conventions. External catalogs also undermine Dagu's attractive low-operations deployment model for smaller teams.
## Use case
A low-operations company-intelligence platform ingests national company registers, parses them into source tables, resolves entities, and publishes canonical and serving datasets. A single source update may affect many downstream datasets, while several workflows may consume the same canonical table.
The operator needs to see that `canonical/companies` was produced from specific source snapshots, determine whether it is stale, trace failures, and trigger only the dependent datasets. The same need applies to ETL/ELT, analytics, ML feature pipelines, search indexing, and data exports.
## Suggested acceptance criteria
- A DAG can declare at least one input and output asset.
- A successful run creates a queryable materialization record with run/step linkage and metadata.
- A failed or aborted producing step does not create a successful materialization.
- Cross-DAG upstream/downstream lineage is queryable through the API and visible in the UI.
- The latest materialization and freshness state are exposed through API, UI, and metrics.
- Existing DAGs, artifacts, and build workflows continue to work unchanged.
Contributor guide
Research direction
Start by reviewing Dagu's existing workflow, artifact, run, API, CLI, UI, and metrics entry points; the issue does not name specific files or tests. Compare those areas with the proposed asset, materialization, lineage, and freshness acceptance criteria. Done means assets are declared and queryable, successful publication is recorded, lineage and freshness are exposed, and existing workflows remain unchanged.
Written by the indexing model from the issue text.
Assessment
- Tech stack
- go
- Domain
- backend, data-engineering, observability
- Issue type
- Feature
- Difficulty
- 5/5
- Estimated time
- Over a week
- Activity status
- Active
- Clarity
- Mostly clear
- Newbie friendliness
- 35/100