EA Mapping, Reimagined — Part 2- A Practical Demo with DuckDB & Python

From documentation to execution: generating SQL from business‑friendly mappings stored as metadata.

What You’ll Build

• A compact metadata schema in DuckDB for sources, targets, and mappings.
• A mapping logic mini-language (plain text) with copy, const, concat, coalesce, and lookup operations.
• A Python generator that reads mappings and emits SQL to materialize a target view/table.
• Lightweight lineage queries and validation checks.

This demo is tool-agnostic. You can drop the SQL into Databricks, run it via Airflow, or compile it into dbt models.

1) DuckDB Metadata Schema

We’ll treat metadata as first-class data.

CREATE SCHEMA IF NOT EXISTS meta;

-- Canonical registry of datasets (schemas, tables, and business names)
CREATE TABLE IF NOT EXISTS meta.dataset (
  dataset_id   INTEGER PRIMARY KEY,
  layer        TEXT,            -- source|integration|exposure
  system_name  TEXT,            -- e.g. crm, sap, s3_raw
  schema_name  TEXT,
  table_name   TEXT,
  business_name TEXT,           -- friendly label
  is_active    BOOLEAN DEFAULT TRUE
);
-- Column definitions per dataset
CREATE TABLE IF NOT EXISTS meta.column (
  dataset_id   INTEGER,
  column_name  TEXT,
  data_type    TEXT,
  business_def TEXT,
  PRIMARY KEY (dataset_id, column_name)
);
-- Mapping units: each row maps one target column using a logic expression
CREATE TABLE IF NOT EXISTS meta.mapping (
  mapping_id    INTEGER PRIMARY KEY,
  target_ds_id  INTEGER,
  target_col    TEXT,
  logic_kind    TEXT,           -- copy|const|concat|coalesce|lookup|sql
  logic_payload TEXT,           -- JSON or simple text (see below)
  precedence    INTEGER DEFAULT 100,  -- combine rules if multiple rows target same column
  note          TEXT
);
-- Optional: reference lookups (for code->meaning or surrogate keys)
CREATE TABLE IF NOT EXISTS meta.lookup (
  lookup_name TEXT,
  key        TEXT,
  value      TEXT
);

Logic conventions

• copy: logic_payload = "source_ds:crm.customers, col:email"
• const: logic_payload = "value:EUROPE"
• concat: logic_payload = "[ {source:'crm.customers', col:'first_name'}, {const:' '}, {source:'crm.customers', col:'last_name'} ]"
• coalesce: logic_payload = "[ {source:'sap.clients', col:'email'}, {source:'crm.customers', col:'email'} ]"
• lookup: logic_payload = "{name:'country_codes', from:'country_code'}" (applies to a previously resolved expression)
• sql: raw SQL fragment when you need total control.

Store logic_payload as JSON strings for safety and parsing.

2) Seed Example Metadata

-- Datasets
INSERT INTO meta.dataset VALUES
  (1,'source','crm','public','customers','CRM Customers',TRUE),
  (2,'source','sap','public','clients','SAP Clients',TRUE),
  (3,'integration','core','public','party','Canonical Party',TRUE);

-- Columns (subset)
INSERT INTO meta.column VALUES
  (1,'customer_id','INTEGER','CRM technical id'),
  (1,'first_name','TEXT','Given name'),
  (1,'last_name','TEXT','Surname'),
  (1,'email','TEXT','Email address'),
  (2,'client_id','INTEGER','SAP technical id'),
  (2,'mail','TEXT','Client email'),
  (2,'country_code','TEXT','ISO alpha-2 code'),
  (3,'party_id','INTEGER','Surrogate id'),
  (3,'full_name','TEXT','Full legal name'),
  (3,'email','TEXT','Preferred email'),
  (3,'country','TEXT','Country name');

-- Lookups
INSERT INTO meta.lookup VALUES
  ('country_codes','NL','Netherlands'),
  ('country_codes','DE','Germany'),
  ('country_codes','BE','Belgium');

-- Mapping rules for integration.core.party
INSERT INTO meta.mapping (mapping_id,target_ds_id,target_col,logic_kind,logic_payload,precedence,note) VALUES
  (1001,3,'full_name','concat','[ {"source":"crm.public.customers","col":"first_name"}, {"const":" "}, {"source":"crm.public.customers","col":"last_name"} ]',100,'Full name from CRM'),
  (1002,3,'email','coalesce','[ {"source":"sap.public.clients","col":"mail"}, {"source":"crm.public.customers","col":"email"} ]',100,'Prefer SAP mail'),
  (1003,3,'country','lookup','{"name":"country_codes","from":"country_code","source":"sap.public.clients"}',100,'Resolve ISO code to name');

In a real project you’d ingest these rows from Notion or CSV.

3) Python Generator (reads metadata → emits SQL)

Below is a compact reference implementation. It:

1. Pulls mapping rows for a target dataset.
2. Builds source CTEs.
3. Resolves each logic_kind into a SQL expression.
4. Produces a final CREATE OR REPLACE VIEW statement.

import json
import duckdb

con = duckdb.connect()
TARGET_DATASET = 3  # core.public.party
# helper to render a source reference "schema.table.col"
def ref(expr):
    src = expr["source"]
    col = expr["col"]
    # quote identifiers minimally
    return f"\"{src}\".\"{col}\"" if "." in src else f"{src}.\"{col}\""
rows = con.execute(
    """
    SELECT target_col, logic_kind, logic_payload, precedence
    FROM meta.mapping
    WHERE target_ds_id = ?
    ORDER BY target_col, precedence
    """,
    [TARGET_DATASET]
).fetchall()
# Group by target column
from collections import defaultdict
by_target = defaultdict(list)
for col, kind, payload, prec in rows:
    by_target[col].append((kind, payload))
expressions = {}
for target_col, items in by_target.items():
    parts = []
    for kind, payload in items:
        if kind == 'copy':
            d = json.loads(payload)
            parts.append(ref(d))
        elif kind == 'const':
            d = json.loads(payload)
            parts.append(f"'{d['value']}'")
        elif kind == 'concat':
            arr = json.loads(payload)
            concat_parts = []
            for e in arr:
                if 'const' in e:
                    concat_parts.append(f"'{e['const']}'")
                else:
                    concat_parts.append(ref(e))
            parts.append(" || ".join(concat_parts))
        elif kind == 'coalesce':
            arr = json.loads(payload)
            coalesce_parts = []
            for e in arr:
                coalesce_parts.append(ref(e))
            parts.append(f"COALESCE({', '.join(coalesce_parts)})")
        elif kind == 'lookup':
            d = json.loads(payload)
            # simple CASE WHEN lookup using meta.lookup
            # note: for production, join a dim table instead
            parts.append(
                f"(SELECT l.value FROM meta.lookup l WHERE l.lookup_name='{d['name']}' AND l.key = {ref({'source': d['source'], 'col': d['from']})} )"
            )
        elif kind == 'sql':
            parts.append(payload)
    # If multiple items per column, wrap last one; precedence allows overrides
    expr = parts[-1]
    expressions[target_col] = expr
# Build FROM sources set by scanning payloads
sources = set()
for items in by_target.values():
    for kind, payload in items:
        if kind in ('concat','coalesce'):
            for e in json.loads(payload):
                if 'source' in e:
                    sources.add(e['source'])
        elif kind in ('copy','lookup'):
            d = json.loads(payload)
            if kind == 'copy':
                sources.add(d['source'])
            if kind == 'lookup':
                sources.add(d['source'])
# naive join list (cross join placeholder). Replace with real join conditions config
from_clause = "\n    ,\n    ".join(sorted(sources))
select_list = ",\n  ".join([f"{expr} AS {col}" for col, expr in expressions.items()])
sql = f"""
CREATE OR REPLACE VIEW core.public.party AS
WITH
  -- Source staging (replace with real filters or SCD2 as needed)
  src AS (
    SELECT * FROM {sorted(sources)[0]}
  )
SELECT
  {select_list}
FROM
  {from_clause};
"""
print(sql)

This generator is intentionally minimal to keep the idea clear. In production you’ll configure joins, SCD2 filters, and null-handling contracts.

4) Example Output (Generated SQL)

For the seeded metadata, a simplified output looks like:

CREATE OR REPLACE VIEW core.public.party AS
SELECT
  "crm.public.customers"."first_name" || ' ' || "crm.public.customers"."last_name" AS full_name,
  COALESCE("sap.public.clients"."mail", "crm.public.customers"."email") AS email,
  (SELECT l.value FROM meta.lookup l WHERE l.lookup_name='country_codes' AND l.key = "sap.public.clients"."country_code") AS country
FROM
  crm.public.customers,
  sap.public.clients;

You’ll replace the implicit cross-join with real join conditions sourced from metadata (e.g., join_key table), or drive it via your Source Access Views.

5) Validation & Lineage Queries

Because everything is metadata, you can add QA with plain SQL.

Find unmapped target columns

SELECT c.column_name
FROM meta.column c
LEFT JOIN (
  SELECT DISTINCT target_col FROM meta.mapping WHERE target_ds_id = 3
) m ON m.target_col = c.column_name
WHERE c.dataset_id = 3 AND m.target_col IS NULL;

Column-level lineage

SELECT m.target_col,
       m.logic_kind,
       m.logic_payload
FROM meta.mapping m
WHERE m.target_ds_id = 3
ORDER BY m.target_col, m.precedence;

6) Extending the Demo

• Joins as metadata: add meta.join with left_ds_id, right_ds_id, condition_sql, join_type and have the generator render proper joins.
• Historization (Dual SCD2): add valid_from, valid_to, and is_current columns in source views; include timepoint filters parameterized in the generator.
• Policies & PII: tag meta.column with pii_level and enforce masking logic in generation.
• dbt integration: render model.sql and schema.yml per target.
• Notion front-end: maintain mapping rows in a Notion database; a sync job writes into meta.* tables.

6a) Dual SCD2 Support (Time-Point & PIT)

Below we extend the metadata and generator to honor Dual SCD2 with time-point filters and optional PIT helpers.

Additional metadata

-- Joins between sources (keyed via SQL condition for flexibility)
CREATE TABLE IF NOT EXISTS meta.join (
  left_ds_id   INTEGER,
  right_ds_id  INTEGER,
  join_type    TEXT DEFAULT 'LEFT',
  condition_sql TEXT
);

-- SCD2 contracts per dataset (applies to its Source Access View)
CREATE TABLE IF NOT EXISTS meta.scd2_contract (
  dataset_id INTEGER,
  valid_from_col TEXT DEFAULT 'valid_from',
  valid_to_col   TEXT DEFAULT 'valid_to',
  is_current_col TEXT DEFAULT 'is_current'
);
-- Optional PIT helper (point-in-time tables) registry
CREATE TABLE IF NOT EXISTS meta.pit (
  pit_name   TEXT,
  dataset_id INTEGER,
  key_cols   TEXT,        -- JSON array of business keys
  asof_col   TEXT         -- reference time column in PIT table
);

Time-point filter pattern

When AS_OF is provided, sources with SCD2 contracts are filtered as:

AS_OF BETWEEN valid_from AND COALESCE(valid_to, TIMESTAMP ‘9999-12-31 23:59:59’)

If a PIT entry exists for a dataset, join the PIT instead of filtering the base view.

Generator delta (Python)

AS_OF = "2025-01-01 00:00:00" # parameterize per run

Fetch SCD2 contracts and joins

scd2 = { (ds,): (vf, vt, ic) for ds, vf, vt, ic in con.execute(
    "SELECT dataset_id, valid_from_col, valid_to_col, is_current_col FROM meta.scd2_contract"
).fetchall() }
joins = con.execute(
    "SELECT left_ds_id, right_ds_id, join_type, condition_sql FROM meta.join"
).fetchall()
# Build FROM with joins rather than cross product
# Map dataset_id->fully qualified name
name_map = { r[0]: f"{r[2]}.{r[3]}.{r[4]}" for r in con.execute(
    "SELECT dataset_id, layer, system_name, schema_name, table_name FROM meta.dataset"
).fetchall() }
# Gather datasets needed from mapping payloads as before -> `needed`
needed = set()
# ... (populate from payloads)
# Build source clauses with SCD2 filters
source_ctes = []
for ds_name in sorted(needed):
    # invert name_map to find dataset_id by name
    ds_id = [k for k,v in name_map.items() if v == ds_name][0]
    base = ds_name
    if (ds_id,) in scd2:
        vf, vt, ic = scd2[(ds_id,)]
        source_ctes.append(
            f"{ds_name.replace('.', '_')}_f AS (\n"
            f"  SELECT * FROM {base}\n"
            f"  WHERE TIMESTAMP '{AS_OF}' BETWEEN {vf} AND COALESCE({vt}, TIMESTAMP '9999-12-31 23:59:59')\n"
            f")"
        )
    else:
        source_ctes.append(f"{ds_name.replace('.', '_')}_f AS (SELECT * FROM {base})")
with_clause = ",\n  ".join(source_ctes)
# Render joins
from_parts = []
if joins:
    # start from the first 'needed' as driving table
    base_name = sorted(needed)[0].replace('.', '_') + "_f"
    sql_from = base_name
    for l, r, jt, cond in joins:
        if name_map[l] in needed and name_map[r] in needed:
            left_alias = name_map[l].replace('.', '_') + "_f"
            right_alias = name_map[r].replace('.', '_') + "_f"
            sql_from += f"\n  {jt} JOIN {right_alias} ON {cond.replace(name_map[l], left_alias).replace(name_map[r], right_alias)}"
    from_sql = sql_from
else:
    from_sql = ",\n  ".join([n.replace('.', '_') + "_f" for n in sorted(needed)])
sql = f"""
CREATE OR REPLACE VIEW core.public.party AS
WITH
  {with_clause}
SELECT
  {select_list}
FROM
  {from_sql};
"""
print(sql)

This keeps SCD2 logic declarative: contracts and joins live in metadata; the generator composes the final query.

6b) dbt Export (model.sql + schema.yml)

Add a small writer to emit the generated SQL as a dbt model and a minimal schema.yml.

from pathlib import Path

project_dir = Path('/mnt/data/dbt_demo')
(project_dir / 'models').mkdir(parents=True, exist_ok=True)
model_path = project_dir / 'models' / 'core__party.sql'
model_path.write_text(sql)
schema_yml = f"""
version: 2
models:
  - name: core__party
    description: Canonical Party view generated from metadata mappings.
    columns:
{chr(10).join([f"      - name: {c}\n        tests: []" for c in expressions.keys()])}
"""
(project_dir / 'models' / 'schema.yml').write_text(schema_yml)
print('dbt assets written to', project_dir)

You can now copy these files into your dbt project (or run from this directory) to operationalize the model.

6c) Notion → DuckDB Sync (one-way)

Keep business edits in Notion and sync into meta.*. Below is a sketch; replace NOTION_SECRET and DATABASE_ID.

import os, requests, json, duckdb

NOTION_SECRET = os.environ.get('NOTION_SECRET')
DATABASE_ID = os.environ.get('NOTION_DB_MAPPINGS')
headers = {
  'Authorization': f'Bearer {NOTION_SECRET}',
  'Notion-Version': '2022-06-28',
  'Content-Type': 'application/json'
}
# pull pages
resp = requests.post(
  f'https://api.notion.com/v1/databases/{DATABASE_ID}/query',
  headers=headers,
  json={"page_size": 100}
)
resp.raise_for_status()
rows = resp.json()['results']
# transform into meta.mapping rows
to_upsert = []
for p in rows:
    props = p['properties']
    target_ds = int(props['Target Dataset']['number'])
    target_col = props['Target Column']['title'][0]['plain_text']
    kind = props['Logic Kind']['select']['name']
    payload = props['Logic Payload']['rich_text'][0]['plain_text']
    prec = int(props.get('Precedence', {}).get('number') or 100)
    note = props.get('Business Note', {}).get('rich_text', [{}])[0].get('plain_text','')
    to_upsert.append((target_ds, target_col, kind, payload, prec, note))
con.execute("BEGIN")
for target_ds, target_col, kind, payload, prec, note in to_upsert:
    con.execute(
      """
      INSERT INTO meta.mapping (mapping_id, target_ds_id, target_col, logic_kind, logic_payload, precedence, note)
      SELECT COALESCE(MAX(mapping_id),0)+1, ?, ?, ?, ?, ?, ? FROM meta.mapping
      """, [target_ds, target_col, kind, payload, prec, note]
    )
con.execute("COMMIT")

Tip: add idempotency (e.g., a _notion_page_id_ column + _MERGE_) and a small scheduler to make sync robust.

7) What This Demonstrates

• Business-friendly mapping lives in metadata, not code.
• The same metadata can generate executable SQL.
• Governance and lineage come for free because the rules are queryable.

Appendix: Suggested Notion Properties for Mappings

• Target Dataset (select)
• Target Column (text)
• Logic Kind (select: copy/const/concat/coalesce/lookup/sql)
• Logic Payload (JSON)
• Precedence (number)
• Business Note (rich text)
• Reviewed By / Approved By (people)

Related Reading

• Part 1 — Rethinking EA Mapping: Building a Business‑Friendly Mapping Capability with DuckDB, Python, and Notion
• Metadata as Data — Controlling and Governing Data Models
• From Historization to Exposure: Building a Usable Data Platform with Dual SCD2