Dependency Mapping & DAG Tracking

In high-throughput ClickHouse analytics pipelines, materialized views (MVs) are not passive query abstractions; they are continuous, insert-driven transformation engines. When a source schema evolves, an ingestion rate spikes, or a cross-table join is introduced without an explicit map of what feeds what, refreshes fire in the wrong order, aggregations go stale, the mutation queue backs up, and the background pool deadlocks. The teams that own this problem — data platform engineers, analytics platform teams, and the Python ETL developers who script the refreshes — need a deterministic, version-controlled Directed Acyclic Graph (DAG) that ties every source table, intermediate view, and analytical sink together. This page covers how to extract that graph from live cluster metadata, resolve a safe execution order, drive it from Python, and keep it in sync as part of materialized view management and sync automation.

The Dependency Graph

A ClickHouse MV chain is a data-flow graph: base MergeTree tables are roots, each MV is an edge that reads one or more upstream objects and writes a target, and the leaf tables are what your BI and data-science consumers query. The moment a chain grows past two hops — raw_events → hourly_agg_mv → daily_summary_mv — implicit ordering assumptions become landmines. A POPULATE on the daily view before the hourly view has caught up produces silently wrong totals; an ALTER TABLE ... MODIFY QUERY on a shared source while a downstream join view is mid-refresh throws Cannot execute DDL.

The goal is to make this graph a first-class, queryable artifact rather than tribal knowledge. Everything downstream — migration ordering, backfill sequencing, incident triage — reads from it.

Metadata Extraction & Graph Construction

ClickHouse does not expose a single system table containing a complete, ready-to-traverse MV dependency graph. Relying solely on the dependencies_table arrays in system.tables yields incomplete lineage, particularly for MVs that reference dictionaries, external tables, or multiple source streams. Production DAG construction requires parsing the create_table_query column and correlating it with system.columns and system.query_log to resolve implicit dependencies.

The extraction pipeline follows three phases:

1. Schema harvesting — Query system.tables filtered to MaterializedView and MergeTree-family engines to isolate transformation targets and base ingestion tables.
2. Query parsing — Extract the SELECT body from create_table_query (AST parsing is safer than regex, but a tokenizer that understands FROM, JOIN, dictGet, and ARRAY JOIN is the minimum) to identify every referenced object.
3. Edge generation — Emit one directed edge (source → target) per referenced object, tagged with an edge type so join edges can be treated differently from direct-feed edges.

-- Production-safe dependency extraction: one row per object in the schema.
-- create_table_query carries the SELECT body we parse for edges.
SELECT
    database,
    name AS view_name,
    create_table_query,
    engine,
    if(engine = 'MaterializedView', 'MV', 'Base') AS object_type
FROM system.tables
WHERE database = 'analytics_prod'
  AND is_temporary = 0
ORDER BY database, name;

Because MVs can be chained, extraction must resolve transitive dependencies and detect cycles before they manifest as Too many parts or Cannot execute DDL at runtime. Validate parsing accuracy against system.query_log: the tables a view actually touched at execution time (tables and columns arrays in the log) catch dynamic references that static DDL parsing can miss.

Core DDL: The Dependency Manifest

Persist the resolved graph in a version-controlled manifest table so the DAG is diffable, auditable, and available to every orchestrator without re-parsing DDL on each run. A ReplacingMergeTree keyed on the edge lets each reconciliation write a fresh snapshot while collapsing to the latest state per edge.

-- The persisted, version-controlled DAG. Rebuilt on every reconciliation run;
-- ReplacingMergeTree collapses repeated edges to the newest discovered_at.
CREATE TABLE analytics_meta.view_dependency_manifest
(
    source_object  LowCardinality(String),          -- upstream table, dictionary, or view
    target_view    LowCardinality(String),          -- MV or target table that consumes it
    edge_type      LowCardinality(String),          -- 'direct' | 'join' | 'dictionary'
    topo_layer     UInt16,                           -- computed topological depth (0 = root)
    manifest_hash  String,                           -- content hash of the resolved DDL, for drift detection
    discovered_at  DateTime64(3, 'UTC') DEFAULT now64(3)
)
ENGINE = ReplacingMergeTree(discovered_at)
PARTITION BY toYYYYMM(discovered_at)                 -- cheap monthly retention of manifest history
ORDER BY (target_view, source_object);               -- one logical row per edge

The manifest_hash column is what makes drift detection cheap: hash the normalized create_table_query of each view and compare against the last stored value. Any mismatch means a view was altered out of band and the graph must be rebuilt. Choosing ReplacingMergeTree here follows the same engine-selection logic covered in the MergeTree engine deep dive — you want last-write-wins semantics on the edge key, not accumulation.

Building and Executing the DAG

The following phases take you from raw metadata to an ordered, executed refresh. Each ends with a verification query so you never advance on faith.

Phase 1 — Harvest and load edges

Run the extraction query, parse each create_table_query, and insert edges into the manifest.

import clickhouse_connect

client = clickhouse_connect.get_client(
    host="clickhouse-prod", database="analytics_prod",
)

rows = client.query(
    """
    SELECT database, name, create_table_query, engine
    FROM system.tables
    WHERE database = 'analytics_prod' AND is_temporary = 0
    """
).result_rows

edges = []
for db, name, ddl, engine in rows:
    for src, kind in parse_referenced_objects(ddl):   # your tokenizer → [(obj, 'direct'|'join'|'dictionary')]
        edges.append((src, name, kind))

client.insert(
    "analytics_meta.view_dependency_manifest",
    edges,
    column_names=["source_object", "target_view", "edge_type"],
)

Verify the edge count landed as expected and no view is missing from the graph:

SELECT count() AS edges, uniqExact(target_view) AS views
FROM analytics_meta.view_dependency_manifest FINAL;

Phase 2 — Resolve topological layers

Load the edges into a graph, reject cycles, and stamp each node’s topological depth back into the manifest.

import networkx as nx

g = nx.DiGraph()
for src, tgt, _ in edges:
    g.add_edge(src, tgt)

if not nx.is_directed_acyclic_graph(g):
    cycle = nx.find_cycle(g)
    raise RuntimeError(f"Cyclic MV chain detected: {cycle}")

# Group nodes into execution layers by longest-path depth from any root.
layers = list(nx.topological_generations(g))

Verify the graph is acyclic and inspect the layering before executing anything:

-- Any target that also appears as its own transitive source is a red flag.
SELECT target_view, groupArray(source_object) AS sources, max(topo_layer) AS layer
FROM analytics_meta.view_dependency_manifest FINAL
GROUP BY target_view
ORDER BY layer;

Phase 3 — Execute layer by layer

Refresh each topological layer in order; within a layer, nodes are independent and can run in parallel up to a bounded concurrency.

python -m dag_sync.run --database analytics_prod --max-concurrency 4

Verify no layer left an MV behind by checking the mutation and merge backlog after each run:

SELECT database, table, count() AS pending
FROM system.mutations
WHERE is_done = 0
GROUP BY database, table
ORDER BY pending DESC;

Topological Resolution & Cross-Table Mapping

Topological sorting determines the safe execution order for schema migrations, POPULATE operations, and incremental backfills. Cross-table dependencies are where ordering gets subtle: an MV that joins users and sessions cannot be safely refreshed if either source is mid-migration. The detailed rules for sequencing join-heavy views after their constituent tables reach a stable schema live in mapping cross-table dependencies for view sync; the summary is that every join edge raises the target’s effective layer to one past the latest of its inputs.

Implement Kahn’s algorithm or use an established graph library to compute layers, and enforce three constraints during resolution:

• Cycle detection — Reject any graph with circular references. ClickHouse MV chains are strictly unidirectional; a circular chain will deadlock the background pool rather than error cleanly.
• Layer grouping — Group nodes by topological depth so disjoint branches execute in parallel while dependent nodes stay serialized.
• Quorum validation — On distributed clusters, confirm insert_quorum and insert_quorum_timeout align with the DAG’s execution window so a layer never commits into a partially-replicated state.

Python ETL Orchestration & State Management

Python is the orchestration layer that turns topological layers into ClickHouse commands. A production sync engine must be idempotent, track state transitions per node, and enforce strict concurrency so parallel layer execution does not starve background merges. Use clickhouse-connect and run blocking client calls off the event loop with asyncio.to_thread.

import asyncio
import clickhouse_connect
import networkx as nx


class DAGSyncEngine:
    def __init__(self, host: str, database: str, max_concurrency: int = 4):
        self.client = clickhouse_connect.get_client(host=host, database=database)
        self.database = database
        self.semaphore = asyncio.Semaphore(max_concurrency)
        self.graph = nx.DiGraph()

    async def execute_layer(self, layer_nodes: list[str]) -> None:
        # Nodes in one topological layer are independent → safe to run together.
        await asyncio.gather(*(self._sync_view(n) for n in layer_nodes))

    async def _sync_view(self, view_name: str) -> None:
        async with self.semaphore:
            # Follow the atomic-swap approach from the creation-patterns guide
            # so a mid-refresh failure never leaves a half-populated target.
            query = f"ALTER TABLE {self.database}.{view_name} MODIFY QUERY ..."
            await asyncio.to_thread(self.client.command, query)

The instantiation and atomic-swap details the _sync_view step relies on come from materialized view creation patterns. When the DAG drives an incremental run rather than a full rebuild, it must inject a watermark filter (WHERE event_ts > :last_sync_ts) or trigger a targeted INSERT ... SELECT backfill; the windowing rules for that come from incremental refresh strategies. Three parameters govern the engine’s behavior under load:

• max_concurrency — Cap at background_pool_size / 2 so refresh threads never crowd out background merges.
• retry_backoff — Exponential backoff (base 2 s, max 30 s) for Too many parts or Lock timeout errors.
• timeout_sec — Set to max_execution_time + 15 s so ClickHouse can gracefully abort a long mutation before the client gives up.

Automated Graph Updates & Operational Visualization

Static DAG definitions rot fast in an active analytics environment. Automated reconciliation keeps the manifest honest: a CI hook or scheduled job diffs the live system.tables state against the stored manifest by comparing manifest_hash per view, and on drift triggers a controlled rebuild instead of a human editing a file. Gate the rebuild behind the same cycle-detection check from Phase 2 so a bad edit can never publish a cyclic graph.

Operational visibility matters just as much. Rendering the DAG as an interactive topology map accelerates incident triage and capacity planning. Export the networkx graph to Graphviz DOT or feed it to a D3 view in an internal dashboard; color nodes by status (ACTIVE, DEGRADED, STALE) and weight edges by throughput so the bottleneck layer is obvious at a glance.

Threshold Tuning & Fallback Resilience

Dependency tracking tells you the safe order; it does not by itself stop degradation under load. ClickHouse processes MV mutations asynchronously through the background pool, and when ingestion spikes the mutation queue in system.mutations saturates, dragging every downstream view behind. The mechanics of that background work are covered in how MergeTree handles background merging; the settings below are the ones that keep a DAG stable while it runs.

Tuning parameters

Fallback chains

Implement recovery paths for failing nodes. If a node fails to sync after three retries, isolate it: detach the MV, flush its queue, and reattach with a simplified query. Route reads to a cached or pre-materialized fallback table until the primary layer stabilizes, so a single stuck view never cascades into the BI or data-science consumers. This mirrors the read-path redundancy patterns in fallback routing and high availability.

Troubleshooting

Cyclic MV chain deadlocks the background pool. Two views reference each other transitively, so neither’s mutations ever drain. Detect it before deploy with the Phase 2 acyclicity check; at runtime, a stuck pool with no progress is the symptom.

SELECT database, table, count() AS stuck_mutations, min(create_time) AS oldest
FROM system.mutations
WHERE is_done = 0
GROUP BY database, table
HAVING oldest < now() - INTERVAL 30 MINUTE;

Fix: KILL MUTATION on the oldest stuck entries, break the cycle by rewriting one view’s query, and rebuild the manifest.

Refresh ordering wrong after an out-of-band ALTER. Someone altered a view without rebuilding the manifest, so the stored topo_layer is stale and a downstream view refreshed before its source.

-- Views whose stored DDL hash no longer matches the live definition.
SELECT name FROM system.tables
WHERE database = 'analytics_prod'
  AND cityHash64(create_table_query) NOT IN (
      SELECT toUInt64OrZero(manifest_hash) FROM analytics_meta.view_dependency_manifest FINAL
  );

Fix: rerun Phase 1–2 reconciliation to regenerate edges and layers before the next execution.

Too many parts during a parallel layer. Concurrency plus max_insert_threads produced more parts than merges can absorb.

SELECT table, count() AS parts
FROM system.parts
WHERE active AND database = 'analytics_prod'
GROUP BY table
ORDER BY parts DESC
LIMIT 10;

Fix: drop max_concurrency to background_pool_size / 2, lower max_insert_threads, and let merges catch up before resuming. The part-count thresholds themselves are tuned in threshold tuning and performance limits.

Stale leaf aggregations with a healthy pool. The pool is idle but a downstream view is behind, usually because a watermark filter excluded late-arriving rows the source only received after the window closed. Confirm by comparing max source timestamp against the view’s last covered window, then trigger a bounded backfill for the affected partitions.

Related

• Materialized view creation patterns — atomic instantiation and swap patterns the DAG engine executes.
• Incremental refresh strategies — watermarking and windowed backfills for DAG-driven refreshes.
• Threshold tuning and performance limits — the part-count and pool thresholds that keep a DAG stable under load.
• Mapping cross-table dependencies for view sync — sequencing join-heavy views after their constituent tables.
• MergeTree engine deep dive — engine semantics behind the manifest and refresh targets.

Up: Materialized View Management & Sync Automation