“

Series: S6 Platform Engineering · Article 9 | Level: Advanced | Reading Time: 18 min

Dual-Layer Data Lineage: Unified Design for Schema-Level and Instance-Level Tracking

#TL;DR

coomia-dip implements a dual-layer data lineage tracking system. Schema-level lineage records definition dependencies between object types and properties, while instance-level lineage records the origin and transformation chains of specific data records. Both layers are managed through a unified LineageGraph model, supporting forward impact analysis ("what will be affected if this field changes") and backward provenance ("how was this value derived"). This article covers the complete design from dual-layer philosophy, graph model implementation, lineage collection, query engine, to visualization integration.

#1. Core Value of Data Lineage

#1.1 Why Dual-Layer Lineage

Single-layer lineage cannot simultaneously serve both architecture governance and data investigation scenarios:

Schema-level lineage: Answers "Which source system fields does Employee.salary come from?" -- supports impact analysis for schema changes
Instance-level lineage: Answers "How was employee John's salary value of $85,000 computed?" -- supports data quality investigation and compliance auditing

Dimension	Schema-Level Lineage	Instance-Level Lineage
Granularity	Object type / Property	Data record / Field value
Change frequency	Low (on schema changes)	High (every data write)
Storage volume	Small (thousands of nodes)	Large (millions of nodes)
Primary users	Architects, governance teams	Data analysts, audit teams
Typical queries	Impact analysis	Data provenance

#1.2 Comparison with Palantir Foundry

Capability	Palantir Foundry	coomia-dip
Schema lineage	Dataset to Transform chain	ObjectType to Property level
Instance lineage	Transaction-level	Record + Field level
Visualization	Monocle	Built-in DAG visualization
Query API	Internal	gRPC + SDK
Storage	Internal	Iceberg + Neo4j

#2. Dual-Layer Lineage Architecture

#2.1 System Architecture

Code

┌────────────────────────────────────────────────────┐
│                Lineage Query API                    │
│           (gRPC + REST, SDK Wrapper)                │
└───────────────────────┬────────────────────────────┘
                        │
┌───────────────────────▼────────────────────────────┐
│              Lineage Graph Engine                   │
│  ┌─────────────────┐  ┌─────────────────────────┐  │
│  │ Schema Lineage  │  │ Instance Lineage        │  │
│  │ (Graph DB)      │  │ (Iceberg + Graph Index) │  │
│  └────────┬────────┘  └────────────┬────────────┘  │
│           │                        │                │
│  ┌────────▼────────────────────────▼────────────┐  │
│  │         Unified Lineage Model                │  │
│  │    (LineageNode, LineageEdge, LineageGraph)   │  │
│  └──────────────────────────────────────────────┘  │
└────────────────────────────────────────────────────┘
                        ▲
┌───────────────────────┼────────────────────────────┐
│           Lineage Collectors                        │
│  ┌──────────┐  ┌──────────┐  ┌─────────────────┐  │
│  │ Schema   │  │ Pipeline │  │ Action          │  │
│  │ Collector│  │ Collector│  │ Collector       │  │
│  └──────────┘  └──────────┘  └─────────────────┘  │
└────────────────────────────────────────────────────┘

#2.2 Unified Lineage Model

Python

class LineageNode(BaseModel):
    """Lineage node"""

    node_id: str = Field(description="Node unique identifier")
    node_type: LineageNodeType = Field(description="Node type")
    layer: LineageLayer = Field(description="Lineage layer")

    name: str
    qualified_name: str  # Fully qualified name (e.g., ObjectType.Property)

    metadata: dict = Field(default_factory=dict)
    classification: ClassificationLevel | None = None

    created_at: datetime
    updated_at: datetime | None = None


class LineageNodeType(str, Enum):
    # Schema level
    OBJECT_TYPE = "object_type"
    PROPERTY = "property"
    LINK_TYPE = "link_type"
    ACTION_TYPE = "action_type"
    EXTERNAL_SOURCE = "external_source"

    # Instance level
    RECORD = "record"
    FIELD_VALUE = "field_value"
    TRANSFORM_EXECUTION = "transform_execution"
    PIPELINE_RUN = "pipeline_run"


class LineageLayer(str, Enum):
    SCHEMA = "schema"
    INSTANCE = "instance"


class LineageEdge(BaseModel):
    """Lineage edge"""

    edge_id: str = Field(default_factory=lambda: str(uuid.uuid4()))
    source_id: str
    target_id: str
    edge_type: LineageEdgeType
    layer: LineageLayer

    transform_type: str | None = None
    transform_expression: str | None = None
    confidence: float = Field(default=1.0, ge=0.0, le=1.0)

    created_at: datetime
    valid_from: datetime | None = None
    valid_until: datetime | None = None


class LineageEdgeType(str, Enum):
    DERIVES_FROM = "derives_from"
    TRANSFORMS_TO = "transforms_to"
    COPIES_FROM = "copies_from"
    AGGREGATES_FROM = "aggregates_from"
    JOINS_WITH = "joins_with"
    REFERENCES = "references"

#3. Schema-Level Lineage

#3.1 Schema Lineage Collector

Python

class SchemaLineageCollector:
    """Schema-level lineage collector"""

    async def on_object_type_created(self, event: SchemaChangeEvent) -> None:
        obj_node = LineageNode(
            node_id=f"schema:{event.schema_id}",
            node_type=LineageNodeType.OBJECT_TYPE,
            layer=LineageLayer.SCHEMA,
            name=event.schema_name,
            qualified_name=event.schema_name,
            created_at=event.timestamp,
        )
        await self._graph.add_node(obj_node)

        for prop in event.properties:
            prop_node = LineageNode(
                node_id=f"schema:{event.schema_id}.{prop.name}",
                node_type=LineageNodeType.PROPERTY,
                layer=LineageLayer.SCHEMA,
                name=prop.name,
                qualified_name=f"{event.schema_name}.{prop.name}",
                classification=prop.classification.level if prop.classification else None,
                created_at=event.timestamp,
            )
            await self._graph.add_node(prop_node)

    async def on_derived_property_defined(
        self,
        object_type: str,
        property_name: str,
        source_expression: str,
        source_properties: list[tuple[str, str]],
    ) -> None:
        target_id = f"schema:{object_type}.{property_name}"

        for src_type, src_prop in source_properties:
            source_id = f"schema:{src_type}.{src_prop}"
            edge = LineageEdge(
                source_id=source_id,
                target_id=target_id,
                edge_type=LineageEdgeType.DERIVES_FROM,
                layer=LineageLayer.SCHEMA,
                transform_type="derive",
                transform_expression=source_expression,
                created_at=datetime.utcnow(),
            )
            await self._graph.add_edge(edge)

#3.2 Schema-Level Queries

Python

class SchemaLineageQuery:
    """Schema-level lineage query"""

    async def get_upstream(
        self, qualified_name: str, max_depth: int = 10,
    ) -> LineageGraph:
        """Get upstream lineage (where data comes from)"""
        return await self._graph.traverse(
            start_node=f"schema:{qualified_name}",
            direction="upstream",
            edge_types=[LineageEdgeType.DERIVES_FROM, LineageEdgeType.COPIES_FROM],
            max_depth=max_depth,
            layer=LineageLayer.SCHEMA,
        )

    async def get_downstream(
        self, qualified_name: str, max_depth: int = 10,
    ) -> LineageGraph:
        """Get downstream lineage (what data affects)"""
        return await self._graph.traverse(
            start_node=f"schema:{qualified_name}",
            direction="downstream",
            edge_types=[LineageEdgeType.TRANSFORMS_TO, LineageEdgeType.DERIVES_FROM],
            max_depth=max_depth,
            layer=LineageLayer.SCHEMA,
        )

    async def impact_analysis(self, qualified_name: str) -> ImpactReport:
        """Change impact analysis"""
        downstream = await self.get_downstream(qualified_name, max_depth=20)
        affected_objects = set()
        affected_properties = set()
        affected_actions = set()

        for node in downstream.nodes:
            if node.node_type == LineageNodeType.OBJECT_TYPE:
                affected_objects.add(node.qualified_name)
            elif node.node_type == LineageNodeType.PROPERTY:
                affected_properties.add(node.qualified_name)
            elif node.node_type == LineageNodeType.ACTION_TYPE:
                affected_actions.add(node.qualified_name)

        return ImpactReport(
            source=qualified_name,
            affected_object_types=list(affected_objects),
            affected_properties=list(affected_properties),
            affected_actions=list(affected_actions),
            total_affected=len(downstream.nodes),
        )

#4. Instance-Level Lineage

#4.1 Instance Lineage Collector

Python

class InstanceLineageCollector:
    """Instance-level lineage collector"""

    async def on_record_created(
        self,
        object_type: str,
        record_id: str,
        source_records: list[SourceRecord] | None = None,
        transform_context: TransformContext | None = None,
    ) -> None:
        record_node = LineageNode(
            node_id=f"instance:{object_type}:{record_id}",
            node_type=LineageNodeType.RECORD,
            layer=LineageLayer.INSTANCE,
            name=record_id,
            qualified_name=f"{object_type}:{record_id}",
            created_at=datetime.utcnow(),
        )
        await self._graph.add_node(record_node)

        if source_records:
            for source in source_records:
                source_id = f"instance:{source.object_type}:{source.record_id}"
                edge = LineageEdge(
                    source_id=source_id,
                    target_id=record_node.node_id,
                    edge_type=LineageEdgeType.DERIVES_FROM,
                    layer=LineageLayer.INSTANCE,
                    transform_type=transform_context.transform_type if transform_context else None,
                    transform_expression=transform_context.expression if transform_context else None,
                    created_at=datetime.utcnow(),
                )
                await self._graph.add_edge(edge)

    async def on_field_computed(
        self,
        object_type: str,
        record_id: str,
        field_name: str,
        computed_value: Any,
        source_fields: list[SourceField],
        computation: str,
    ) -> None:
        target_id = f"instance:{object_type}:{record_id}:{field_name}"

        field_node = LineageNode(
            node_id=target_id,
            node_type=LineageNodeType.FIELD_VALUE,
            layer=LineageLayer.INSTANCE,
            name=f"{field_name}={computed_value}",
            qualified_name=f"{object_type}:{record_id}.{field_name}",
            metadata={"value": str(computed_value), "computation": computation},
            created_at=datetime.utcnow(),
        )
        await self._graph.add_node(field_node)

        for source in source_fields:
            source_id = f"instance:{source.object_type}:{source.record_id}:{source.field_name}"
            edge = LineageEdge(
                source_id=source_id,
                target_id=target_id,
                edge_type=LineageEdgeType.DERIVES_FROM,
                layer=LineageLayer.INSTANCE,
                transform_type="compute",
                transform_expression=computation,
                created_at=datetime.utcnow(),
            )
            await self._graph.add_edge(edge)

#4.2 Instance-Level Queries

Python

class InstanceLineageQuery:
    """Instance-level lineage query"""

    async def trace_value_origin(
        self, object_type: str, record_id: str, field_name: str,
    ) -> ValueOriginTrace:
        """Trace the origin of a field value"""
        start_node = f"instance:{object_type}:{record_id}:{field_name}"
        graph = await self._graph.traverse(
            start_node=start_node,
            direction="upstream",
            max_depth=50,
            layer=LineageLayer.INSTANCE,
        )

        steps = []
        for edge in graph.edges_in_order():
            steps.append(TraceStep(
                source=edge.source_id,
                target=edge.target_id,
                transform=edge.transform_expression,
                timestamp=edge.created_at,
            ))

        return ValueOriginTrace(
            field=f"{object_type}.{field_name}",
            record_id=record_id,
            steps=steps,
            leaf_sources=[n for n in graph.leaf_nodes()],
        )

    async def get_record_provenance(
        self, object_type: str, record_id: str,
    ) -> RecordProvenance:
        """Get complete provenance of a record"""
        node_id = f"instance:{object_type}:{record_id}"
        graph = await self._graph.traverse(
            start_node=node_id,
            direction="upstream",
            max_depth=20,
            layer=LineageLayer.INSTANCE,
        )

        return RecordProvenance(
            record_id=record_id,
            object_type=object_type,
            source_count=len(graph.leaf_nodes()),
            transform_count=len(graph.edges),
            lineage_graph=graph,
        )

#5. Lineage Storage

#5.1 Hybrid Storage Strategy

Python

class HybridLineageStore:
    """Hybrid storage - Graph DB for schema, Iceberg for instances"""

    def __init__(
        self,
        graph_store: GraphStore,       # Neo4j / JanusGraph
        table_store: IcebergStore,     # Iceberg tables
    ):
        self._graph = graph_store      # Schema-level + hot instance-level
        self._table = table_store      # Cold instance-level

    async def add_node(self, node: LineageNode) -> None:
        if node.layer == LineageLayer.SCHEMA:
            await self._graph.upsert_node(node)
        else:
            await self._graph.upsert_node(node)
            await self._table.append_node(node)

    async def add_edge(self, edge: LineageEdge) -> None:
        if edge.layer == LineageLayer.SCHEMA:
            await self._graph.upsert_edge(edge)
        else:
            await self._graph.upsert_edge(edge)
            await self._table.append_edge(edge)

    async def archive_old_instances(self, before: datetime) -> int:
        """Archive old instance lineage (remove from graph, keep in Iceberg)"""
        return await self._graph.delete_nodes(
            layer=LineageLayer.INSTANCE,
            created_before=before,
        )

#6. Lineage and Security Integration

#6.1 Classification Propagation

Lineage relationships drive automatic classification propagation:

Python

class LineageDrivenClassification:
    """Lineage-driven classification propagation"""

    async def propagate_classification(
        self, source_node_id: str, new_classification: ClassificationLevel,
    ) -> list[str]:
        downstream = await self._lineage_query.get_downstream(
            source_node_id, max_depth=20,
        )

        affected_nodes = []
        for node in downstream.nodes:
            if node.classification is None or node.classification < new_classification:
                node.classification = new_classification
                await self._graph.update_node(node)
                affected_nodes.append(node.node_id)

        return affected_nodes

#6.2 Audit Integration

Lineage query operations are themselves recorded in the audit trail:

Python

@audit_tracked(AuditEventType.DATA_ACCESS, action="lineage_query")
async def trace_value_origin(self, object_type, record_id, field_name):
    ...

#7. Testing Strategy

Python

class TestDualLayerLineage:
    async def test_schema_lineage_creation(self):
        collector = SchemaLineageCollector(graph)
        await collector.on_derived_property_defined(
            object_type="Employee",
            property_name="total_compensation",
            source_expression="base_salary + bonus",
            source_properties=[
                ("Employee", "base_salary"),
                ("Employee", "bonus"),
            ],
        )

        upstream = await query.get_upstream("Employee.total_compensation")
        assert len(upstream.nodes) == 3
        assert len(upstream.edges) == 2

    async def test_instance_lineage_trace(self):
        collector = InstanceLineageCollector(graph)
        await collector.on_field_computed(
            object_type="Employee",
            record_id="emp-001",
            field_name="total_compensation",
            computed_value=85000,
            source_fields=[
                SourceField("Employee", "emp-001", "base_salary"),
                SourceField("Employee", "emp-001", "bonus"),
            ],
            computation="base_salary + bonus",
        )

        trace = await query.trace_value_origin("Employee", "emp-001", "total_compensation")
        assert len(trace.steps) == 2
        assert len(trace.leaf_sources) == 2

    async def test_impact_analysis(self):
        report = await query.impact_analysis("SourceSystem.raw_salary")
        assert "Employee.base_salary" in report.affected_properties
        assert "Employee.total_compensation" in report.affected_properties

#8. Production Best Practices

#8.1 Storage Strategy

Lineage Layer	Hot Storage	Warm Storage	Cold Storage
Schema-level	Permanent (Graph DB)	-	-
Instance-level	30 days (Graph index)	365 days (Iceberg)	5 years (archive)

#8.2 Performance Optimization

Schema-level queries: < 50ms (graph index)
Instance-level provenance: < 200ms (hot data)
Batch lineage collection: async pipeline, non-blocking to write path
Periodically archive old instance data to keep graph index lightweight

#8.3 Data Quality

Lineage completeness checks: periodically verify all derived properties have lineage records
Orphan node detection: periodically clean up lineage nodes with no edge connections
Lineage gap alerting: trigger alerts when data transformations fail to produce lineage records

#9. Summary

The coomia-dip dual-layer data lineage system covers all lineage requirements from architecture governance to data investigation through schema-level and instance-level tracking. Key design highlights:

Dual-layer design: Schema-level for impact analysis, instance-level for data provenance
Unified model: Both layers share LineageNode/LineageEdge models, reducing complexity
Hybrid storage: Graph database + Iceberg balances query performance and storage economics
Security integration: Lineage-driven classification propagation with deep audit system integration
Extensible: Collector plugin mechanism supports new data transformation scenarios

The next article will explore coomia-dip's historical data replay capabilities.