Postgres-to-Postgres pipelines

Source (public.tablename)
    ↓
Normalisation   ← rename/exclude columns (Normalisation tab)
    ↓
Raw layer  (raw.public__tablename)
    ↓
dbt Layer       ← SQL model using {{ source('raw', 'public__tablename') }} (dbt Layer tab)
    ↓
Destination  (analytics.your_model)

ALTER TABLE analytics.your_table ADD PRIMARY KEY (id);

SELECT *
FROM {{ source }}

SELECT
  id,
  email,
  first_name,
  last_name,
  created_at,
  updated_at
FROM {{ source }}

SELECT
  id,
  usr_email         AS email,
  usr_created_ts    AS created_at,
  usr_updated_ts    AS updated_at,
  acct_status       AS status
FROM {{ source }}

SELECT
  id,
  smallint_col::INTEGER       AS int_val,
  integer_col::BIGINT         AS big_val,
  real_col::DOUBLE PRECISION  AS float_val,
  numeric_col::NUMERIC(18,4)  AS precise_val,
  char_col::TEXT              AS char_text,
  varchar_col::TEXT           AS varchar_text,
  timestamp_col::TIMESTAMPTZ  AS ts_utc,
  date_col::TIMESTAMP         AS date_as_ts
FROM {{ source }}

SELECT *
FROM {{ source }}
WHERE is_deleted = false
  AND status IN ('active', 'pending')
  AND created_at >= '2024-01-01'

SELECT
  id,
  first_name || ' ' || last_name  AS full_name,
  UPPER(email)                    AS email_upper,
  EXTRACT(YEAR FROM created_at)   AS signup_year,
  CURRENT_TIMESTAMP               AS loaded_at,
  CASE
    WHEN score >= 90 THEN 'high'
    WHEN score >= 60 THEN 'medium'
    ELSE 'low'
  END                             AS score_tier
FROM {{ source }}

SELECT
  id,
  order_number,
  CAST(amount_cents AS NUMERIC) / 100.0  AS amount_usd,
  CAST(tax_cents    AS NUMERIC) / 100.0  AS tax_usd,
  CAST(total_cents  AS NUMERIC) / 100.0  AS total_usd,
  currency,
  created_at
FROM {{ source }}

SELECT
  id,
  payload->>'event_type'              AS event_type,
  (payload->>'amount')::NUMERIC       AS amount,
  (payload->>'user_id')::UUID         AS user_id,
  (payload->>'timestamp')::TIMESTAMPTZ AS event_at,
  payload->'metadata'                 AS metadata_jsonb
FROM {{ source }}

SELECT
  customer_id,
  COUNT(*)          AS order_count,
  SUM(total_amount) AS lifetime_value,
  AVG(total_amount) AS avg_order_value,
  MIN(created_at)   AS first_order_at,
  MAX(created_at)   AS last_order_at
FROM {{ source }}
GROUP BY customer_id

SELECT
  id,
  user_id,
  status,
  status = 'active'  AS is_active,
  status = 'churned' AS is_churned,
  status = 'trial'   AS is_trial,
  created_at
FROM {{ source }}

SELECT
  id,
  int4range_col::TEXT   AS int_range,
  int8range_col::TEXT   AS bigint_range,
  numrange_col::TEXT    AS num_range,
  tsrange_col::TEXT     AS ts_range,
  tstzrange_col::TEXT   AS tstz_range,
  daterange_col::TEXT   AS date_range
FROM {{ source }}

SELECT
  id,
  point_col::TEXT   AS point,
  polygon_col::TEXT AS polygon,
  circle_col::TEXT  AS circle,
  box_col::TEXT     AS box
FROM {{ source }}

-- Land as JSONB
SELECT
  id,
  to_jsonb(int_array_col)  AS int_array_json,
  to_jsonb(text_array_col) AS text_array_json
FROM {{ source }}

-- Unnest into one row per element
SELECT
  id,
  unnest(text_array_col) AS tag
FROM {{ source }}

SELECT
  id,
  COALESCE(preferred_name, first_name, 'Unknown')  AS display_name,
  COALESCE(mobile_phone, home_phone, work_phone)   AS contact_phone,
  COALESCE(updated_at, created_at)                 AS last_seen_at
FROM {{ source }}

SELECT
  id,
  smallint_col::BIGINT         AS widened_int,
  real_col::DOUBLE PRECISION   AS widened_float,
  char_col::TEXT               AS widened_char,
  varchar_col::TEXT            AS widened_varchar,
  date_col::TIMESTAMPTZ        AS widened_date,
  timestamp_col::TIMESTAMPTZ   AS widened_ts,
  inet_col::TEXT               AS inet_text,
  uuid_col::TEXT               AS uuid_text,
  bool_col::TEXT               AS bool_text,
  jsonb_col::TEXT              AS json_text
FROM {{ source }}