Tutorial: Migrate Redshift Warehouse to Fabric / Databricks SQL¶

-- Connect to Redshift via psql
-- psql -h acme-analytics.abc123.us-gov-west-1.redshift.amazonaws.com -U admin -d analytics -p 5439

-- Cluster node configuration
SELECT
  node_type,
  number_of_nodes
FROM stv_wlm_service_class_config
LIMIT 1;

-- Database sizes
SELECT
  trim(pgdb.datname) AS database_name,
  ROUND(SUM(b.mbytes)::NUMERIC / 1024, 2) AS size_gb
FROM stv_tbl_perm a
JOIN pg_database pgdb ON pgdb.oid = a.db_id
JOIN (
  SELECT tbl, COUNT(*) AS mbytes
  FROM stv_blocklist
  GROUP BY tbl
) b ON a.id = b.tbl
GROUP BY 1
ORDER BY 2 DESC;

-- Top 20 most expensive queries (by execution time) in the last 30 days
SELECT
  trim(querytxt) AS query_text,
  COUNT(*) AS execution_count,
  AVG(total_exec_time) / 1000000 AS avg_seconds,
  MAX(total_exec_time) / 1000000 AS max_seconds,
  SUM(total_exec_time) / 1000000 AS total_seconds
FROM stl_query
WHERE starttime > DATEADD(day, -30, GETDATE())
  AND querytxt NOT LIKE 'padb_fetch_sample%'
  AND querytxt NOT LIKE 'COPY%'
  AND querytxt NOT LIKE 'UNLOAD%'
GROUP BY 1
ORDER BY total_seconds DESC
LIMIT 20;

-- WLM queue utilization
SELECT
  service_class,
  trim(name) AS queue_name,
  num_queued_queries,
  num_executing_queries,
  avg_queue_time / 1000000 AS avg_queue_seconds
FROM stl_wlm_query
JOIN stv_wlm_service_class_config USING (service_class)
WHERE starttime > DATEADD(day, -7, GETDATE())
GROUP BY 1, 2, 3, 4, 5
ORDER BY 1;

-- Table scan frequency (identifies hot tables)
SELECT
  trim(s.perm_table_name) AS table_name,
  COUNT(*) AS scan_count,
  AVG(s.rows_pre_filter) AS avg_rows_scanned,
  AVG(s.rows) AS avg_rows_returned
FROM stl_scan s
WHERE s.starttime > DATEADD(day, -30, GETDATE())
  AND s.perm_table_name NOT LIKE 'stl_%'
  AND s.perm_table_name NOT LIKE 'pg_%'
GROUP BY 1
ORDER BY 2 DESC
LIMIT 30;

-- All user tables with row counts, sizes, and distribution/sort keys
SELECT
  trim(n.nspname) AS schema_name,
  trim(c.relname) AS table_name,
  t.diststyle,
  trim(d.colname) AS dist_key,
  trim(sk.attname) AS sort_key,
  t.tbl_rows::BIGINT AS row_count,
  t.size AS size_mb,
  CASE WHEN t.unsorted > 0 THEN t.unsorted ELSE 0 END AS unsorted_pct
FROM svv_table_info t
JOIN pg_class c ON t.table_id = c.oid
JOIN pg_namespace n ON c.relnamespace = n.oid
LEFT JOIN (
  SELECT attrelid, trim(attname) AS colname
  FROM pg_attribute
  WHERE attisdistkey = true
) d ON c.oid = d.attrelid
LEFT JOIN pg_attribute sk ON c.oid = sk.attrelid AND sk.attsortkeyord = 1
WHERE n.nspname NOT IN ('pg_catalog', 'information_schema', 'pg_internal')
ORDER BY t.size DESC;

-- Generate DDL for all tables in a schema
SELECT
  ddl
FROM admin.v_generate_tbl_ddl
WHERE schemaname = 'public'
ORDER BY tablename, seq;

-- If admin.v_generate_tbl_ddl is not available, use pg_dump:
-- pg_dump -h <host> -U admin -d analytics --schema-only --no-owner -t 'public.*' > redshift_schema.sql

CREATE TABLE public.fact_orders (
  order_id       BIGINT       IDENTITY(1,1) NOT NULL,
  customer_id    INTEGER      NOT NULL ENCODE lzo,
  order_date     DATE         NOT NULL,
  product_id     INTEGER      NOT NULL ENCODE lzo,
  quantity       SMALLINT     NOT NULL ENCODE raw,
  unit_price     DECIMAL(10,2) NOT NULL ENCODE az64,
  gross_amount   DECIMAL(12,2) NOT NULL ENCODE az64,
  region         VARCHAR(64)  NOT NULL ENCODE bytedict,
  order_status   VARCHAR(32)  NOT NULL ENCODE bytedict,
  created_at     TIMESTAMP    DEFAULT getdate() ENCODE az64
)
DISTKEY(customer_id)
COMPOUND SORTKEY(order_date, region);

CREATE TABLE IF NOT EXISTS analytics_prod.gold.fact_orders (
  order_id       BIGINT       GENERATED ALWAYS AS IDENTITY,
  customer_id    INT          NOT NULL,
  order_date     DATE         NOT NULL,
  product_id     INT          NOT NULL,
  quantity       SMALLINT     NOT NULL,
  unit_price     DECIMAL(10,2) NOT NULL,
  gross_amount   DECIMAL(12,2) NOT NULL,
  region         STRING       NOT NULL,
  order_status   STRING       NOT NULL,
  created_at     TIMESTAMP    DEFAULT current_timestamp()
)
USING DELTA
PARTITIONED BY (order_date)
COMMENT 'Migrated from Redshift public.fact_orders'
TBLPROPERTIES (
  'delta.autoOptimize.optimizeWrite' = 'true',
  'delta.autoOptimize.autoCompact' = 'true',
  'quality' = 'gold',
  'source_system' = 'redshift_migration'
);

-- Apply Z-ordering after initial data load
-- OPTIMIZE analytics_prod.gold.fact_orders ZORDER BY (customer_id, region);

-- UNLOAD each table to S3 in Parquet format
UNLOAD ('SELECT * FROM public.fact_orders')
TO 's3://acme-migration-staging/redshift-export/fact_orders/'
IAM_ROLE 'arn:aws-us-gov:iam::123456789:role/redshift-unload-role'
FORMAT AS PARQUET
PARALLEL ON
MAXFILESIZE 256 MB
ALLOWOVERWRITE;

-- For large tables, UNLOAD with partition predicate to parallelize
UNLOAD ('SELECT * FROM public.fact_orders WHERE order_date >= ''2024-01-01''')
TO 's3://acme-migration-staging/redshift-export/fact_orders/2024/'
IAM_ROLE 'arn:aws-us-gov:iam::123456789:role/redshift-unload-role'
FORMAT AS PARQUET
PARALLEL ON
MAXFILESIZE 256 MB;

UNLOAD ('SELECT * FROM public.fact_orders WHERE order_date < ''2024-01-01''')
TO 's3://acme-migration-staging/redshift-export/fact_orders/pre-2024/'
IAM_ROLE 'arn:aws-us-gov:iam::123456789:role/redshift-unload-role'
FORMAT AS PARQUET
PARALLEL ON
MAXFILESIZE 256 MB;

-- Enable auto-optimize for ongoing writes
ALTER TABLE analytics_prod.gold.fact_orders SET TBLPROPERTIES (
  'delta.autoOptimize.optimizeWrite' = 'true',
  'delta.autoOptimize.autoCompact' = 'true',
  'delta.columnMapping.mode' = 'name',
  'delta.minReaderVersion' = '2',
  'delta.minWriterVersion' = '5'
);

-- Enable change data feed for downstream CDC consumers
ALTER TABLE analytics_prod.gold.fact_orders SET TBLPROPERTIES (
  'delta.enableChangeDataFeed' = 'true'
);

CREATE OR REPLACE PROCEDURE sp_daily_revenue_summary(run_date DATE)
AS $$
BEGIN
  DELETE FROM reporting.daily_revenue WHERE report_date = run_date;

  INSERT INTO reporting.daily_revenue
  SELECT
    run_date AS report_date,
    p.category,
    r.region,
    COUNT(DISTINCT o.customer_id) AS unique_customers,
    SUM(o.quantity) AS total_units,
    SUM(o.gross_amount) AS total_revenue,
    AVG(o.gross_amount) AS avg_order_value,
    GETDATE() AS generated_at
  FROM public.fact_orders o
  JOIN public.dim_products p ON o.product_id = p.product_id
  JOIN public.dim_regions r ON o.region = r.region_code
  WHERE o.order_date = run_date
  GROUP BY 1, 2, 3;
END;
$$ LANGUAGE plpgsql;

-- models/gold/daily_revenue_summary.sql
{{ config(
    materialized='incremental',
    unique_key=['report_date', 'category', 'region'],
    incremental_strategy='merge',
    partition_by=['report_date']
) }}

SELECT
  o.order_date                        AS report_date,
  p.category,
  r.region,
  COUNT(DISTINCT o.customer_id)       AS unique_customers,
  SUM(o.quantity)                     AS total_units,
  SUM(o.gross_amount)                 AS total_revenue,
  AVG(o.gross_amount)                 AS avg_order_value,
  current_timestamp()                 AS generated_at
FROM {{ ref('stg_fact_orders') }} o
JOIN {{ ref('stg_dim_products') }} p ON o.product_id = p.product_id
JOIN {{ ref('stg_dim_regions') }} r  ON o.region = r.region_code
{% if is_incremental() %}
WHERE o.order_date >= date_sub(current_date(), 3)
{% endif %}
GROUP BY 1, 2, 3