SQL Templating

# Internal implementation (for reference)
from jinja2.sandbox import SandboxedEnvironment
from jinja2 import StrictUndefined
 
env = SandboxedEnvironment(undefined=StrictUndefined)

ref('layer.name')            → 2-part (same namespace)
ref('namespace.layer.name')  → 3-part (cross-namespace)

-- 2-part: references bronze.raw_orders in the current namespace
SELECT * FROM {{ ref('bronze.raw_orders') }}
 
-- 3-part: references a table in a different namespace
SELECT * FROM {{ ref('shared.silver.customers') }}

-- 2-part resolves to (example):
SELECT * FROM iceberg_scan('s3://rat/ecommerce/bronze/raw_orders/metadata/v1.metadata.json')
 
-- 3-part resolves to:
SELECT * FROM iceberg_scan('s3://rat/shared/silver/customers/metadata/v1.metadata.json')

landing_zone('zone_name')

SELECT *
FROM read_csv_auto(
    '{{ landing_zone("orders") }}',
    header = true,
    all_varchar = true
)

SELECT *
FROM read_csv_auto(
    's3://rat/ecommerce/landing/orders/**',
    header = true,
    all_varchar = true
)

s3://{bucket}/{namespace}/landing/{zone_name}/**

# Normal execution
s3://rat/ecommerce/landing/orders/**

# Preview mode
s3://rat/ecommerce/landing/orders/_samples/**

-- @merge_strategy: incremental
-- @unique_key: user_id
 
SELECT * FROM {{ ref('bronze.raw_users') }}
{% if is_incremental() %}
WHERE updated_at > (SELECT MAX(updated_at) FROM {{ this }})
{% endif %}

SELECT * FROM iceberg_scan('s3://rat/ecommerce/bronze/raw_users/metadata/...')
WHERE updated_at > (SELECT MAX(updated_at) FROM iceberg_scan('s3://rat/ecommerce/silver/dim_users/metadata/...'))

SELECT
    *,
    '{{ run_started_at }}' AS _loaded_at
FROM {{ ref('bronze.raw_events') }}

SELECT
    *,
    '2026-02-16T09:30:00Z' AS _loaded_at
FROM iceberg_scan('...')

-- @merge_strategy: incremental
-- @unique_key: order_id
-- @watermark_column: updated_at
 
SELECT * FROM {{ ref('bronze.raw_orders') }}
{% if is_incremental() %}
WHERE updated_at > '{{ watermark_value }}'
{% endif %}

-- @merge_strategy: incremental
-- @unique_key: id
-- @watermark_column: updated_at
 
SELECT * FROM {{ ref('bronze.raw_data') }}
{% if is_incremental() %}
WHERE updated_at > '{{ watermark_value }}'
{% endif %}

-- @merge_strategy: scd2
-- @unique_key: customer_id
-- @scd_valid_from: valid_from
-- @scd_valid_to: valid_to
 
SELECT
    customer_id,
    name,
    email,
    tier,
    {% if is_scd2() %}
    CURRENT_TIMESTAMP AS valid_from,
    NULL AS valid_to,
    {% endif %}
    updated_at
FROM {{ ref('bronze.raw_customers') }}

-- @merge_strategy: snapshot
-- @partition_column: snapshot_date
 
SELECT
    *,
    {% if is_snapshot() %}
    CURRENT_DATE AS snapshot_date,
    {% endif %}
FROM {{ ref('silver.inventory') }}

-- @description: Clean and deduplicate customer records
-- @merge_strategy: incremental
-- @unique_key: customer_id
-- @watermark_column: updated_at
 
SELECT * FROM {{ ref('bronze.raw_customers') }}
{% if is_incremental() %}
WHERE updated_at > '{{ watermark_value }}'
{% endif %}

# @description: Fetch data from external REST API
# @merge_strategy: full_refresh
# @max_retries: 3
# @retry_delay_seconds: 30
 
import requests
 
def run(context):
    response = requests.get("https://api.example.com/data")
    return response.json()

-- @merge_strategy: incremental    ← extracted
-- @unique_key: id                 ← extracted
-- This is a regular comment       ← NOT extracted (no @key pattern), extraction stops
 
SELECT * FROM {{ ref('bronze.data') }}
-- @description: ignored           ← NOT extracted (below SQL, too late)

{% if is_incremental() %}
WHERE updated_at > '{{ watermark_value }}'
{% endif %}
 
{% if is_scd2() %}
-- SCD2 specific columns
{% elif is_snapshot() %}
-- Snapshot specific columns
{% else %}
-- Default columns
{% endif %}

{# This is a Jinja comment — it will not appear in the compiled SQL #}
SELECT * FROM {{ ref('bronze.data') }}

SELECT
    id,
    name
    {%- if is_scd2() -%}
    , valid_from
    , valid_to
    {%- endif %}
FROM {{ ref('bronze.data') }}

-- @description: Orders enriched with customer data, incrementally processed
-- @merge_strategy: incremental
-- @unique_key: order_id
-- @watermark_column: updated_at
-- @materialized: table
 
{# Join orders with customers and only process new records #}
SELECT
    o.order_id,
    o.customer_id,
    o.status,
    o.total_amount,
    o.currency,
    o.order_date,
    o.updated_at,
    c.name AS customer_name,
    c.segment AS customer_segment,
    c.country AS customer_country,
    '{{ run_started_at }}' AS _processed_at
FROM {{ ref('bronze.raw_orders') }} o
LEFT JOIN {{ ref('silver.customers') }} c
    ON o.customer_id = c.customer_id
WHERE o.order_id IS NOT NULL
{% if is_incremental() %}
  AND o.updated_at > '{{ watermark_value }}'
{% endif %}