Lineage Tracing

What you’ll learn¶

Trace backward to find an artifact’s sources
Trace forward for impact analysis
Use one-hop vs transitive queries
Bulk DataFrame traversal with walk_forward and walk_backward

The Provenance Graphs tutorial showed how to visualize lineage. This tutorial covers the programmatic side: querying the provenance store to trace ancestors, descendants, and lineage chains in code.

Prerequisites: Provenance Graphs, Exploring Results.
Estimated time: 15 minutes
GPU required: No.

from __future__ import annotations

import polars as pl

from artisan.operations.curator import Filter
from artisan.operations.examples import (
    DataGenerator,
    DataTransformer,
    MetricCalculator,
)
from artisan.orchestration import Backend, PipelineManager
from artisan.provenance import walk_backward, walk_forward
from artisan.storage import ArtifactStore
from artisan.utils import tutorial_setup
from artisan.visualization import inspect_pipeline

Build a sample pipeline¶

We’ll build the same five-step pipeline from the Provenance Graphs tutorial: generate, transform, score, filter, refine. This gives us a rich provenance graph with branching and filtering to query.

env = tutorial_setup("lineage_tracing")

pipeline = PipelineManager.create(
    name="lineage_tutorial",
    delta_root=env.delta_root,
    staging_root=env.staging_root,
    working_root=env.working_root,
)
output = pipeline.output

pipeline.run(
    operation=DataGenerator,
    name="generate",
    params={"count": 5, "seed": 42},
    backend=Backend.LOCAL,
)
pipeline.run(
    operation=DataTransformer,
    name="transform",
    inputs={"dataset": output("generate", "datasets")},
    backend=Backend.LOCAL,
)
pipeline.run(
    operation=MetricCalculator,
    name="score",
    inputs={"dataset": output("transform", "dataset")},
    backend=Backend.LOCAL,
)
pipeline.run(
    operation=Filter,
    name="filter",
    inputs={"passthrough": output("transform", "dataset")},
    params={
        "criteria": [
            {"metric": "distribution.median", "operator": "gt", "value": 0.3},
        ]
    },
    backend=Backend.LOCAL,
)
pipeline.run(
    operation=DataTransformer,
    name="refine",
    inputs={"dataset": output("filter", "passthrough")},
    backend=Backend.LOCAL,
)

pipeline.finalize()
inspect_pipeline(env.delta_root)

store = ArtifactStore(env.delta_root)

Backward tracing: where did this come from?¶

The most common lineage question: given an artifact, what are its sources? The ProvenanceStore (accessed via store.provenance) provides both one-hop and transitive ancestor queries.

Full ancestry chain¶

get_ancestor_ids walks the entire provenance graph backward from a given artifact, returning every ancestor reachable via any number of hops.

# Pick a metric from step 2
metric_ids = store.provenance.load_artifact_ids_by_type("metric", step_numbers=[2])
metric_id = sorted(metric_ids)[0]
print(f"Starting from metric: {metric_id[:12]}... (step 2)")

# Get ALL ancestors (transitive)
ancestor_ids = store.provenance.get_ancestor_ids(metric_id)
type_map = store.provenance.load_type_map(ancestor_ids)
step_map = store.provenance.load_step_map(set(ancestor_ids))

print(f"\nFull ancestry ({len(ancestor_ids)} ancestors):")
for aid in sorted(ancestor_ids, key=lambda x: step_map.get(x, 99)):
    print(f"  step {step_map[aid]}: {type_map[aid]} {aid[:12]}...")

Starting from a metric at step 2, the ancestry chain goes: metric (step 2) <- dataset (step 1) <- dataset (step 0). The step-0 dataset is a source artifact with no ancestors.

You can also filter ancestors by type:

# Only get data ancestors (skip metrics, file_refs, etc.)
data_ancestors = store.provenance.get_ancestor_ids(metric_id, ancestor_type="data")
print(f"Data ancestors only: {len(data_ancestors)}")
for aid in data_ancestors:
    step = store.provenance.get_artifact_step_number(aid)
    print(f"  step {step}: {aid[:12]}...")

One-hop: direct parents¶

When you only need the immediate parent(s), use get_direct_ancestors. This returns only the artifacts one step back in the provenance graph.

# Direct parent of the metric
parents = store.provenance.get_direct_ancestors(metric_id)
parent_id = parents[0]
parent_type = store.provenance.load_type_map([parent_id])[parent_id]
parent_step = store.provenance.get_artifact_step_number(parent_id)
print(f"Direct parent: {parent_id[:12]}... (type={parent_type}, step={parent_step})")

# Direct parent of the parent
grandparents = store.provenance.get_direct_ancestors(parent_id)
if grandparents:
    gp_id = grandparents[0]
    gp_step = store.provenance.get_artifact_step_number(gp_id)
    print(f"Grandparent:   {gp_id[:12]}... (step={gp_step})")

# Source artifacts have no ancestors
source_parents = store.provenance.get_direct_ancestors(gp_id)
print(f"Source has ancestors: {bool(source_parents)}")

Forward tracing: what was derived from this?¶

The reverse question: given a source artifact, find everything downstream. This is essential for impact analysis — if you discover an issue with a source, forward tracing tells you exactly which downstream results are affected.

# Pick a source dataset from step 0
source_ids = store.provenance.load_artifact_ids_by_type("data", step_numbers=[0])
source_id = sorted(source_ids)[0]
print(f"Starting from source: {source_id[:12]}... (step 0)")

# Get ALL descendants (transitive)
all_descendants = store.provenance.get_descendant_ids(source_id)
type_map = store.provenance.load_type_map(all_descendants)
step_map = store.provenance.load_step_map(set(all_descendants))

print(f"\nAll descendants ({len(all_descendants)}):")
for aid in sorted(all_descendants, key=lambda x: step_map.get(x, 99)):
    print(f"  step {step_map[aid]}: {type_map[aid]} {aid[:12]}...")

Typed forward queries¶

Filter descendants by type to answer specific questions like “what metrics were computed from this source?”

# Only metric descendants
metric_descendants = store.provenance.get_descendant_ids(
    source_id, descendant_type="metric"
)
print(f"Metric descendants: {len(metric_descendants)}")

# Only data descendants
data_descendants = store.provenance.get_descendant_ids(
    source_id, descendant_type="data"
)
print(f"Data descendants: {len(data_descendants)}")

One-hop queries¶

get_direct_descendants returns only the immediate children of a set of source artifacts. Unlike get_descendant_ids (which walks the full graph), this stays at one level.

It accepts a set of source IDs and returns a dict mapping each source to its children — useful for batch lookups.

# What was directly derived from each step-0 source?
descendant_map = store.provenance.get_direct_descendants(source_ids)

for src_id, children in sorted(descendant_map.items(), key=lambda x: x[0]):
    child_types = store.provenance.load_type_map(children)
    print(f"Source {src_id[:12]}... ->")
    for child in children:
        child_step = store.provenance.get_artifact_step_number(child)
        print(f"    step {child_step}: {child_types[child]} {child[:12]}...")

Each step-0 source has exactly one direct descendant at step 1 (the transformed dataset). The metrics at step 2 are descendants of the step-1 datasets, not direct descendants of step 0.

You can also filter one-hop descendants by type:

# Direct metric descendants of step-1 datasets
step1_ids = store.provenance.load_artifact_ids_by_type("data", step_numbers=[1])
metric_children = store.provenance.get_direct_descendants(
    step1_ids, target_artifact_type="metric"
)
print(f"Step-1 datasets with metric children: {len(metric_children)}")

Bulk traversal with DataFrames¶

For batch operations across many artifacts, use walk_forward and walk_backward from artisan.provenance. These operate on Polars DataFrames and are efficient for large-scale traversal.

First, load the provenance edges as a DataFrame:

# Load all edges spanning steps 0 through 4
edges = store.provenance.load_edges_df(0, 4, include_target_type=True)
print(f"Total provenance edges: {edges.height}")
edges.head(5)

Walking forward¶

walk_forward takes a DataFrame of source artifact IDs and walks through the edges to find all reachable descendants. Optionally filter by target type.

# Start from all step-0 sources
sources_df = pl.DataFrame({"artifact_id": sorted(source_ids)})

# Find all metric descendants of step-0 sources
metrics_reached = walk_forward(sources_df, edges, target_type="metric")
print(f"Metrics reached from {sources_df.height} sources: {metrics_reached.height}")
metrics_reached.head(5)

Walking backward¶

walk_backward takes candidate artifacts and target artifacts, and finds which candidates trace back to which targets. This is useful for matching — e.g., “which metrics at step 2 came from which sources at step 0?”

# Which step-2 metrics trace back to which step-0 sources?
candidates = pl.DataFrame({"artifact_id": sorted(metric_ids)})
targets = pl.DataFrame({"artifact_id": sorted(source_ids)})

matched = walk_backward(candidates, targets, edges)
print(f"Matched {matched.height} (metric, source) pairs")
matched

Practical example: investigating a suspect artifact¶

A common workflow: you spot a questionable artifact and need to understand whether the issue is isolated or systematic.

Scenario: A final artifact at step 4 looks wrong. Trace backward to its source, then forward from that source to find all siblings — other artifacts derived from the same source.

# Pick a final artifact from step 4
step4_ids = store.provenance.load_artifact_ids_by_type("data", step_numbers=[4])
suspect_id = sorted(step4_ids)[0]
print(f"Suspect artifact: {suspect_id[:12]}... (step 4)")

# Trace backward to its source
ancestors = store.provenance.get_ancestor_ids(suspect_id, ancestor_type="data")
step_map = store.provenance.load_step_map(set(ancestors))
source = [a for a in ancestors if step_map[a] == 0][0]
print(f"Original source: {source[:12]}... (step 0)")

# Now trace forward from that source to find ALL derived artifacts
siblings = store.provenance.get_descendant_ids(source)
sib_types = store.provenance.load_type_map(siblings)
sib_steps = store.provenance.load_step_map(set(siblings))

print(f"\nAll artifacts from the same source ({len(siblings)} total):")
for aid in sorted(siblings, key=lambda x: sib_steps.get(x, 99)):
    marker = " <-- suspect" if aid == suspect_id else ""
    print(f"  step {sib_steps[aid]}: {sib_types[aid]} {aid[:12]}...{marker}")

If the suspect and its siblings all look wrong, the issue is likely in the source data. If only the suspect is wrong, the issue is in a later transformation step.

Which method for which question¶

Question	Method	Returns
What are this artifact’s immediate parents?	`store.provenance.get_direct_ancestors(id)`	`list[str]`
What are the immediate children of these sources?	`store.provenance.get_direct_descendants(ids)`	`dict[str, list[str]]`
What is the full ancestry of this artifact?	`store.provenance.get_ancestor_ids(id)`	`list[str]`
What is derived from this artifact (full tree)?	`store.provenance.get_descendant_ids(id)`	`list[str]`
Same, but only metrics/data?	Add `ancestor_type=` or `descendant_type=`	`list[str]`
Batch: which sources produced these targets?	`walk_backward(candidates, targets, edges)`	`DataFrame`
Batch: what downstream artifacts from these sources?	`walk_forward(sources, edges, target_type=)`	`DataFrame`
Load raw provenance edges	`store.provenance.load_edges_df(min, max)`	`DataFrame`

Summary¶

You learned to query lineage programmatically at three levels:

One-hop queries — get_direct_ancestors and get_direct_descendants for immediate parent/child lookups.
Transitive queries — get_ancestor_ids and get_descendant_ids for full lineage chains.
Bulk DataFrame walks — walk_forward and walk_backward for batch traversal across many artifacts.

Next steps¶

Interactive Filter — Explore metrics and set filter thresholds interactively
Provenance Graphs — Visualize the lineage you’re querying
Provenance System — How the dual provenance model works under the hood
Inspecting Provenance — Task-oriented recipes for common provenance queries