We observed that executing a “lift and shift” of legacy, sprawling SQL scripts onto an enterprise cloud data warehouse fails to resolve core structural data issues. It transitions architectural technical debt into a variable, unconstrained operational expense. Moving unoptimized queries onto infinite-compute cloud platforms masks underlying engineering deficiencies rather than fixing them. We reject this pattern. We run a decentralized data architecture using dbt-core paired with optimized PostgreSQL instances to enforce localized data governance, strict structural boundaries, and code-driven engineering principles directly at the storage layer.
The 4-Layer Blueprint: Explicit Separation of Concerns
We enforce a strict 4-layer Directed Acyclic Graph (DAG) that isolates transformation steps into distinct, single-responsibility schemas. We do not allow short-circuiting across these boundaries. Models can only select from the immediate layer beneath them or external sources mapped specifically to that schema.
graph TD
A[Raw Ingest Tables] --> B[Staging Layer: stg_]
B --> C[Intermediate Layer: int_]
C --> D[Core Layer: dim_ / fct_]
D --> E[Datamarts Layer: mrt_]1. Staging Layer (stg_)
This layer maintains a strict 1:1 mapping with raw source tables. We enforce a zero-join constraint here to prevent premature complexity. The staging layer executes basic datatype casting, standardizes cryptic source column naming conventions to uniform snake_case, and filters out corrupt bytes or null-padded primary keys. We run these models exclusively as views to avoid unnecessary disk I/O and write amplification penalties on the PostgreSQL storage engine.
By default, PostgreSQL folds views directly into downstream queries. This means the staging layer introduces zero runtime storage overhead while establishing a clean schema abstraction layer that insulates downstream models from upstream source mutations.
2. Intermediate Layer (int_)
This layer handles heavy intra-domain processing. We isolate state mutations, complex window functions, and multi-table joins inside this layer to keep downstream presentation layers clean. Intermediate models are completely hidden from external consumer business units. We materialize these models as ephemeral blocks or physical tables depending on code complexity and execution frequency.
Using ephemeral materialization compiles the model as a Common Table Expression (CTE) in downstream operations. This reduces disk footprints but can increase memory pressure on the PostgreSQL work_mem allocation when processing large volumes. When a transformation exceeds our allocated memory threshold per operation, we enforce a physical table materialization strategy to prevent the query planner from spilling sorting operations to disk (pg_stat_database.temp_files).
3. Core Layer (core_)
The domain’s backbone and enterprise source of truth. We deploy standardized dimension tables (dim_) representing structural enterprise entities alongside immutable fact tables (fct_) capturing high-velocity events. Data inside this layer is completely clean, fully typed, and structured for long-term storage and cross-domain compatibility. We avoid highly normalized entity-relationship models here, choosing instead optimized star schemas that minimize join depths during consumer execution.
4. Datamarts Layer (mrt_)
The explicit consumption layer. Built directly on top of the Core Layer, these models pre-join wide arrays, compute hyper-specific aggregation logic, and expose specialized semantic layers tailored to business groups or outbound LLM interfaces. Outside domains are strictly prohibited from penetrating past this layer when pulling metrics.
Hands-On Implementation: Code Blueprints
We translate this 4-layer model into production-grade dbt-core and SQL structures running inside our localized media_consumption domain database.
Layer 1: The Staging Model (models/staging/stg_media_events.sql)
We configure this model to run strictly as a view. No multi-table transformations or aggregations are permitted.
SQL
{{ config(
materialized='view',
schema='staging'
) }}
with source as (
select * from {{ source('raw_ingest', 'streaming_logs') }}
),
renamed as (
select
cast(event_id as varchar(64)) as event_id,
cast(user_hash as varchar(64)) as user_id,
cast(content_id as varchar(32)) as content_id,
cast(device_type as varchar(32)) as device_category,
cast(duration_seconds as integer) as playback_duration_seconds,
cast(created_at as timestamp) as occurred_at
from source
where event_id is not null
)
select * from renamed
Layer 2: The Intermediate Model (models/intermediate/int_media_events_sessionized.sql)
We introduce session identification markers by processing raw sequences via window functions. Materializing this layer as a transient table balances the cost of I/O against the CPU strain of recalculating deep window logic downstream.
SQL
{{ config(
materialized='table',
schema='intermediate'
) }}
with staging_events as (
select * from {{ ref('stg_media_events') }}
),
time_deltas as (
select
event_id,
user_id,
content_id,
device_category,
playback_duration_seconds,
occurred_at,
lag(occurred_at) over (
partition by user_id
order by occurred_at asc
) as previous_event_at
from staging_events
),
session_flags as (
select
*,
case
when extract(epoch from (occurred_at - previous_event_at)) / 60 > 30 then 1
when previous_event_at is null then 1
else 0
end as is_new_session
from time_deltas
),
session_counters as (
select
*,
sum(is_new_session) over (
partition by user_id
order by occurred_at asc
) as session_sequence_number
from session_flags
)
select
event_id,
user_id,
content_id,
device_category,
playback_duration_seconds,
occurred_at,
md5(concat(user_id, '-', session_sequence_number::text)) as session_id
from session_counters
Layer 3: The Core Layer (models/core/fct_playback_sessions.sql)
We enforce incremental materialization strategies for our fact tables to optimize disk performance and eliminate full-table scans. We use a surrogate key strategy computed via deterministic hashing. To ensure high scan efficiency on time-series records without the overhead of heavy B-Tree maintenance, we deploy Block Range Indexes (BRIN) on the date field.
SQL
{{ config(
materialized='incremental',
unique_key='playback_session_pk',
schema='core',
indexes=[
{'columns': ['activity_date'], 'type': 'brin'},
{'columns': ['user_id'], 'type': 'btree'}
]
) }}
with sessionized_events as (
select * from {{ ref('int_media_events_sessionized') }}
{% if is_incremental() %}
where occurred_at >= (select max(activity_date) from {{ this }}) - interval '2 days'
{% endif %}
),
aggregated_sessions as (
select
date_trunc('day', occurred_at)::date as activity_date,
session_id,
user_id,
content_id,
device_category,
count(distinct event_id) as total_playback_events,
sum(playback_duration_seconds) as total_viewtime_seconds
from sessionized_events
where playback_duration_seconds > 0
group by 1, 2, 3, 4, 5
)
select
md5(concat(activity_date::text, '-', session_id)) as playback_session_pk,
activity_date,
session_id,
user_id,
content_id,
device_category,
total_playback_events,
total_viewtime_seconds
from aggregated_sessions
Layer 4: The Datamarts Layer (models/datamarts/mrt_marketing_weekly_engagement.sql)
This layer presents pre-aggregated data directly to downstream metrics pipelines, operational services, and localized domain querying mechanisms.
SQL
{{ config(
materialized='table',
schema='datamarts'
) }}
with core_sessions as (
select * from {{ ref('fct_playback_sessions') }}
),
core_users as (
select * from {{ ref('dim_users') }}
),
weekly_rollup as (
select
date_trunc('week', s.activity_date)::date as playback_week,
u.subscription_tier,
u.country_code,
sum(s.total_playback_events) as weekly_events,
sum(s.total_viewtime_seconds) as weekly_viewtime_seconds
from core_sessions s
join core_users u on s.user_id = u.user_id
group by 1, 2, 3
)
select * from weekly_rollup
Scaling Beyond the Monolith: Fitting the 4-Layer Stack into an Enterprise Data Mesh
Implementing a true enterprise Data Mesh requires moving away from a centralized, single-warehouse team and transitioning toward decentralized, domain-oriented data ownership. Based on frameworks for enterprise distributed architectures, our Data Mesh treats data strictly as a product.
The mistake many organizations make is assuming that decentralized data products require massive independent cloud infrastructures for every single department. Instead, we run an isolated Data Mesh by treating independent PostgreSQL database instances (or isolated micro-schemas inside a central cluster) as standalone Data Products, orchestrated independently via modular dbt-core repositories.
graph TD
subgraph ENTERPRISE DATA MESH
MC[Media Consumption Data Product <br> Postgres + dbt-core]
MKT[Marketing Data Product <br> Postgres + dbt-core]
FIN[Finance Data Product <br> Postgres + dbt-core]
end
MC --> AI[Cross-Mesh AI Routing Layer <br> PydanticAI]
MKT --> AI
FIN --> AIIn this architecture, the 4-layer layout serves as the internal engineering engine inside each individual data product node:
- Internal Privacy (Staging & Intermediate): Staging and Intermediate layers remain fully internal to the domain’s data product team. Outside teams cannot query them, isolating upstream system changes from breaking downstream dependencies.
- The Domain Contract (Core & Datamarts): The Core and Datamarts layers represent the clean, version-controlled Data Product Interface. When another domain (e.g., Finance) needs to consume streaming engagement metrics, they cross-query only the exposed Datamart or Core models via controlled, read-only database shares or schema linkages.
Leveraging Data Mesh Documentation for Agentic AI
Structuring our analytics platform into clear Core and Datamart layers wrapped in rich declarative metadata unlocks an invaluable enterprise asset: making our distributed Data Mesh completely discoverable and navigable by LLM Agents.
In a standard Data Mesh, humans struggle with discovery—they have to scroll through complex data catalogs just to find out which team owns what dataset. Agentic AI platforms face an even greater hurdle when trying to execute Text-to-SQL cross-domain queries without explicit boundaries. By thoroughly documenting our layers inside dbt’s schema.yml configuration files, we create a decentralized semantic blueprint that turns each separate domain node into an indexable data asset.
The Domain Interface Blueprint: models/datamarts/schema.yml
We embed operational context, ownership boundaries, and direct descriptive prompts inside our configuration configurations. This allows AI agents to read the schema format as an immutable software contract.
YAML
version: 2
models:
- name: mrt_marketing_weekly_engagement
description: >
[DOMAIN: MEDIA CONSUMPTION] Use this data product model ONLY when users ask for high-level
marketing metrics, aggregated weekly view times, or content consumption trends segmented by
subscription tiers. Outbound point-of-contact: @media-data-team.
columns:
- name: playback_week
description: "The starting date (Monday) of the aggregation week."
data_tests:
- not_null
- name: subscription_tier
description: "The customer's billing level. Common values are 'free', 'basic', or 'premium'."
data_tests:
- not_null
- name: weekly_viewtime_seconds
description: "The total duration of media consumed in seconds during this week for this segment."
Federated Metadata Architecture for Autonomous Mesh Discovery
When we compile or run our distributed data nodes, dbt-core generates a localized artifact: target/manifest.json. This single file contains the entire structured graph of that domain’s data ecosystem—every model description, every explicit data type, and every constraint.
In our automated architecture, these manifest.json files are pulled from their decentralized repositories and consolidated into a lightweight, central metadata storage directory.
graph LR
A[Domain A: Manifest] --> D[Federated AI
Semantic Index]
B[Domain B: Manifest] --> D
C[Domain C: Manifest] --> D
D --> E[Multi-Agent Router]When an enterprise AI agent (built on advanced orchestration frameworks like PydanticAI) receives a natural language query—such as “Compare our weekly video streaming view times in Germany against our ad-spend marketing budgets from last week”—the system routes the query autonomously across the mesh:
- Mesh Discovery: The Agent scans the consolidated semantic index and immediately identifies that video view times live inside the
media_consumptiondatabase node undermrt_marketing_weekly_engagement, while ad budgets live inside thefinancedomain database undermrt_marketing_ad_spend. - Deterministic SQL Compilation: Because both distributed schemas are explicitly documented with data types, parameters, and localized keys, the AI constructs an accurate, federated PostgreSQL query joining across both domain data products on the first attempt:
SQL
select
v.playback_week,
v.weekly_viewtime_seconds,
f.actual_ad_spend_eur
from media_consumption.mrt_marketing_weekly_engagement v
join finance.mrt_marketing_ad_spend f
on v.playback_week = f.budget_week
and v.country_code = f.target_country_code
where v.country_code = 'DE'
and v.playback_week = '2026-06-22';
The Multi-Agent Mesh Router via PydanticAI
We parse these compiled manifest definitions and load them directly into our AI orchestration workflows. The agent evaluates system tasks, matches the semantic context against our declared dbt metadata, and generates deterministic SQL targeting the proper database nodes.
Python
import json
from pydantic import BaseModel, Field
from pydantic_ai import Agent, RunContext
class MeshQueryRouter(BaseModel):
target_database: str = Field(description="The specific domain PostgreSQL instance connection string identifier.")
target_schema: str = Field(description="The isolated target layer schema name, typically core or datamarts.")
compiled_sql: str = Field(description="The complete executable ANSI SQL query targeting the domain.")
with open("target/manifest.json", "r") as f:
dbt_manifest_data = json.load(f)
semantic_context_block = {}
for node_id, node_meta in dbt_manifest_data.get("nodes", {}).items():
if node_meta.get("config", {}).get("schema") in ["core", "datamarts"]:
semantic_context_block[node_id] = {
"name": node_meta.get("name"),
"description": node_meta.get("description"),
"columns": {k: v.get("description") for k, v in node_meta.get("columns", {}).items()}
}
mesh_routing_agent = Agent(
'openai:gpt-4o',
result_type=MeshQueryRouter,
system_prompt=(
"You are an expert data engineer routing requests across an enterprise PostgreSQL Data Mesh. "
"Analyze the provided dbt semantic architecture block to locate the correct domain data asset. "
"Generate strict, production-ready ANSI SQL based on column descriptions. "
"Never query internal staging (stg_) or intermediate (int_) schemas. "
f"Available Semantic Context Blueprint: {json.dumps(semantic_context_block)}"
)
)
@mesh_routing_agent.tool
def verify_table_existence(ctx: RunContext[None], target_table: str) -> bool:
"""Internal validation tool to check structural alignment within metadata boundaries."""
return any(target_table in node for node in semantic_context_block.keys())
user_prompt = "Calculate our total weekly media viewing hours inside Germany for premium subscribers"
agent_execution_result = mesh_routing_agent.run_sync(user_prompt)
Next Phase Execution: Operationalizing Cross-Node Schema Migrations and Replication Topology
Our next deployment architecture addresses the operational stability of decentralized PostgreSQL nodes when scaling to concurrent multi-region clusters. We reject manual schema synchronizations across distinct domain repositories; instead, we deploy deterministic automation tools to enforce consistency across the network mesh.
We will expose the engineering blueprints and raw configurations for:
- Automated DDL Schema State Synchronizers: Utilizing specialized CI/CD pipelines to parse dbt manifest state files, auto-generating migration scripts that execute across remote Foreign Data Wrapper endpoints without locking active consumer tables.
- Replication Slot Monitoring and Failover State Machines: Setting up autonomous daemons and Prometheus triggers to identify replication lag within logical replication pipelines, forcing programmatic failover routes before WAL storage directories exhaust localized disk arrays.
- Distributed Connection Pooling Architecture: Optimizing multi-domain pgBouncer configurations to handle connection spikes from cross-mesh multi-agent routing engines without exhausting native database engine processes.
- Native pg_vector Semantic Vector Generation: Embedding localized text-to-vector embedding generation pipelines directly within the datamart schemas, removing external network latency from the AI semantic routing loop.
