setup_databricks.sql

-- =============================================================================
-- Medallion architecture for the OpenMetadata + Kafka demo
-- Run this in your Databricks Free Edition SQL Editor after creating the
-- Unity Catalog storage credential + external location for the S3 staging bucket.
-- =============================================================================

CREATE CATALOG IF NOT EXISTS demo;
USE CATALOG demo;

CREATE SCHEMA IF NOT EXISTS bronze COMMENT 'Raw CDC events from Kafka (append-only)';
CREATE SCHEMA IF NOT EXISTS silver COMMENT 'Cleaned current-state tables';
CREATE SCHEMA IF NOT EXISTS gold   COMMENT 'Business aggregates for analytics';

-- ---------------------------------------------------------------------------
-- Bronze: Landing tables for the Confluent Delta Lake Sink.
-- Connector settings delta.lake.catalog / delta.lake.database target demo.bronze
-- with short table names (customers, products, ...). auto.create is off —
-- create these tables before starting the sink.
--
-- Schema MUST match Debezium ExtractNewRecordState + staging Avro:
--   - INT ids (not BIGINT for PKs)
--   - TIMESTAMPTZ → STRING (io.debezium.time.ZonedTimestamp ISO-8601)
--   - __op / __table / __source_ts_ms / __deleted (unwrap rewrite adds __deleted)
--   - no partition column (staging files omit it)
--   - order_items has no line_total (Postgres GENERATED; compute in silver)
-- If COPY INTO fails with schema mismatch, run databricks/recreate_bronze.sql
-- ---------------------------------------------------------------------------

CREATE TABLE IF NOT EXISTS demo.bronze.customers (
  id             INT,
  email          STRING,
  full_name      STRING,
  phone          STRING,
  created_at     STRING,
  __op           STRING,
  __table        STRING,
  __source_ts_ms BIGINT,
  __deleted      STRING
) USING DELTA
COMMENT 'Raw customer CDC events from Kafka topic dbz.ecommerce.public.customers';

CREATE TABLE IF NOT EXISTS demo.bronze.products (
  id             INT,
  sku            STRING,
  name           STRING,
  category       STRING,
  unit_price     STRING,
  updated_at     STRING,
  __op           STRING,
  __table        STRING,
  __source_ts_ms BIGINT,
  __deleted      STRING
) USING DELTA
COMMENT 'Raw product CDC events from Kafka topic dbz.ecommerce.public.products';

CREATE TABLE IF NOT EXISTS demo.bronze.orders (
  id             INT,
  customer_id    INT,
  status         STRING,
  currency       STRING,
  total_amount   STRING,
  placed_at      STRING,
  updated_at     STRING,
  __op           STRING,
  __table        STRING,
  __source_ts_ms BIGINT,
  __deleted      STRING
) USING DELTA
COMMENT 'Raw order CDC events from Kafka topic dbz.ecommerce.public.orders';

CREATE TABLE IF NOT EXISTS demo.bronze.order_items (
  id             INT,
  order_id       INT,
  product_id     INT,
  quantity       INT,
  unit_price     STRING,
  __op           STRING,
  __table        STRING,
  __source_ts_ms BIGINT,
  __deleted      STRING
) USING DELTA
COMMENT 'Raw order_item CDC events from Kafka topic dbz.ecommerce.public.order_items';

-- ---------------------------------------------------------------------------
-- (Optional legacy path) If an older sink run landed in workspace.default.*,
-- promote into demo.bronze — normally unnecessary once catalog/database are set.
-- ---------------------------------------------------------------------------
-- INSERT INTO demo.bronze.customers SELECT * FROM workspace.default.customers;
-- INSERT INTO demo.bronze.products  SELECT * FROM workspace.default.products;
-- INSERT INTO demo.bronze.orders    SELECT * FROM workspace.default.orders;
-- INSERT INTO demo.bronze.order_items SELECT * FROM workspace.default.order_items;

-- ---------------------------------------------------------------------------
-- Silver: current-state tables (dedupe CDC append log → latest row per PK).
-- Temporal columns: ZonedTimestamp ISO-8601 strings → TIMESTAMP via to_timestamp().
-- Re-run after the sink has flushed data.
-- ---------------------------------------------------------------------------

CREATE OR REPLACE TABLE demo.silver.customers AS
SELECT id, email, full_name, phone,
       to_timestamp(created_at) AS created_at,
       __source_ts_ms AS last_cdc_ts
FROM (
  SELECT *,
         ROW_NUMBER() OVER (PARTITION BY id ORDER BY __source_ts_ms DESC) AS rn
  FROM demo.bronze.customers
  WHERE COALESCE(__op, 'c') <> 'd'
    AND COALESCE(__deleted, 'false') <> 'true'
)
WHERE rn = 1;

CREATE OR REPLACE TABLE demo.silver.products AS
SELECT id, sku, name, category,
       CAST(unit_price AS DECIMAL(12, 2)) AS unit_price,
       to_timestamp(updated_at) AS updated_at,
       __source_ts_ms AS last_cdc_ts
FROM (
  SELECT *,
         ROW_NUMBER() OVER (PARTITION BY id ORDER BY __source_ts_ms DESC) AS rn
  FROM demo.bronze.products
  WHERE COALESCE(__op, 'c') <> 'd'
    AND COALESCE(__deleted, 'false') <> 'true'
)
WHERE rn = 1;

CREATE OR REPLACE TABLE demo.silver.orders AS
SELECT id, customer_id, status, currency,
       CAST(total_amount AS DECIMAL(12, 2)) AS total_amount,
       to_timestamp(placed_at) AS placed_at,
       to_timestamp(updated_at) AS updated_at,
       __source_ts_ms AS last_cdc_ts
FROM (
  SELECT *,
         ROW_NUMBER() OVER (PARTITION BY id ORDER BY __source_ts_ms DESC) AS rn
  FROM demo.bronze.orders
  WHERE COALESCE(__op, 'c') <> 'd'
    AND COALESCE(__deleted, 'false') <> 'true'
)
WHERE rn = 1;

CREATE OR REPLACE TABLE demo.silver.order_items AS
SELECT id, order_id, product_id, quantity,
       CAST(unit_price AS DECIMAL(12, 2)) AS unit_price,
       quantity * CAST(unit_price AS DECIMAL(12, 2)) AS line_total,
       __source_ts_ms AS last_cdc_ts
FROM (
  SELECT *,
         ROW_NUMBER() OVER (PARTITION BY id ORDER BY __source_ts_ms DESC) AS rn
  FROM demo.bronze.order_items
  WHERE COALESCE(__op, 'c') <> 'd'
    AND COALESCE(__deleted, 'false') <> 'true'
)
WHERE rn = 1;

-- ---------------------------------------------------------------------------
-- Gold: analytics aggregates (Unity Catalog records lineage from silver → gold)
-- ---------------------------------------------------------------------------

CREATE OR REPLACE TABLE demo.gold.customer_order_summary AS
SELECT
  c.id AS customer_id,
  c.email,
  c.full_name,
  COUNT(DISTINCT o.id) AS order_count,
  COALESCE(SUM(o.total_amount), 0) AS lifetime_value,
  MAX(o.placed_at) AS last_order_at
FROM demo.silver.customers c
LEFT JOIN demo.silver.orders o
  ON c.id = o.customer_id
 AND o.status <> 'cancelled'
GROUP BY c.id, c.email, c.full_name;

CREATE OR REPLACE TABLE demo.gold.daily_revenue AS
SELECT
  CAST(o.placed_at AS DATE) AS order_date,
  o.currency,
  COUNT(DISTINCT o.id) AS orders,
  SUM(oi.line_total) AS revenue
FROM demo.silver.orders o
JOIN demo.silver.order_items oi ON o.id = oi.order_id
WHERE o.status <> 'cancelled'
GROUP BY CAST(o.placed_at AS DATE), o.currency;

CREATE OR REPLACE TABLE demo.gold.product_performance AS
SELECT
  p.id AS product_id,
  p.sku,
  p.name,
  p.category,
  SUM(oi.quantity) AS units_sold,
  SUM(oi.line_total) AS revenue
FROM demo.silver.products p
LEFT JOIN demo.silver.order_items oi ON p.id = oi.product_id
LEFT JOIN demo.silver.orders o
  ON oi.order_id = o.id
 AND o.status <> 'cancelled'
GROUP BY p.id, p.sku, p.name, p.category;