Databricks Dbsql

BEGIN  DECLARE v_count INT;  DECLARE v_status STRING DEFAULT 'pending';
  SET v_count = (SELECT COUNT(*) FROM catalog.schema.raw_orders WHERE status = 'new');
  IF v_count > 0 THEN    INSERT INTO catalog.schema.processed_orders    SELECT *, current_timestamp() AS processed_at    FROM catalog.schema.raw_orders    WHERE status = 'new';
    SET v_status = 'completed';  ELSE    SET v_status = 'skipped';  END IF;
  SELECT v_status AS result, v_count AS rows_processed;END

CREATE OR REPLACE PROCEDURE catalog.schema.upsert_customers(  IN p_source STRING,  OUT p_rows_affected INT)LANGUAGE SQLSQL SECURITY INVOKERBEGIN  DECLARE EXIT HANDLER FOR SQLEXCEPTION  BEGIN    SET p_rows_affected = -1;    SIGNAL SQLSTATE '45000'      SET MESSAGE_TEXT = concat('Upsert failed for source: ', p_source);  END;
  MERGE INTO catalog.schema.dim_customer AS t  USING (SELECT * FROM identifier(p_source)) AS s  ON t.customer_id = s.customer_id  WHEN MATCHED THEN UPDATE SET *  WHEN NOT MATCHED THEN INSERT *;
  SET p_rows_affected = (SELECT COUNT(*) FROM identifier(p_source));END;
-- Invoke:CALL catalog.schema.upsert_customers('catalog.schema.staging_customers', ?);

CREATE OR REPLACE MATERIALIZED VIEW catalog.schema.daily_revenue  CLUSTER BY (order_date)  SCHEDULE EVERY 1 HOUR  COMMENT 'Hourly-refreshed daily revenue by region'AS SELECT    order_date,    region,    SUM(amount) AS total_revenue,    COUNT(DISTINCT customer_id) AS unique_customersFROM catalog.schema.fact_ordersJOIN catalog.schema.dim_store USING (store_id)GROUP BY order_date, region;

-- Traditional SQL rewritten with pipe syntaxFROM catalog.schema.fact_orders  |> WHERE order_date >= current_date() - INTERVAL 30 DAYS  |> AGGREGATE SUM(amount) AS total, COUNT(*) AS cnt GROUP BY region, product_category  |> WHERE total > 10000  |> ORDER BY total DESC  |> LIMIT 20;

-- Classify support ticketsSELECT  ticket_id,  description,  ai_classify(description, ARRAY('billing', 'technical', 'account', 'feature_request')) AS category,  ai_analyze_sentiment(description) AS sentimentFROM catalog.schema.support_ticketsLIMIT 100;
-- Extract entities from textSELECT  doc_id,  ai_extract(content, ARRAY('person_name', 'company', 'dollar_amount')) AS entitiesFROM catalog.schema.contracts;
-- General-purpose AI query with structured outputSELECT ai_query(  'databricks-meta-llama-3-3-70b-instruct',  concat('Summarize this customer feedback in JSON with keys: topic, sentiment, action_items. Feedback: ', feedback),  returnType => 'STRUCT<topic STRING, sentiment STRING, action_items ARRAY<STRING>>') AS analysisFROM catalog.schema.customer_feedbackLIMIT 50;

-- Find stores within 5km of each customer using H3 indexingWITH customer_h3 AS (  SELECT *, h3_longlatash3(longitude, latitude, 7) AS h3_cell  FROM catalog.schema.customers),store_h3 AS (  SELECT *, h3_longlatash3(longitude, latitude, 7) AS h3_cell  FROM catalog.schema.stores)SELECT  c.customer_id,  s.store_id,  ST_Distance(    ST_Point(c.longitude, c.latitude),    ST_Point(s.longitude, s.latitude)  ) AS distance_mFROM customer_h3 cJOIN store_h3 s ON h3_ischildof(c.h3_cell, h3_toparent(s.h3_cell, 5))WHERE ST_Distance(  ST_Point(c.longitude, c.latitude),  ST_Point(s.longitude, s.latitude)) < 5000;

-- Create table with case-insensitive collationCREATE TABLE catalog.schema.products (  product_id BIGINT GENERATED ALWAYS AS IDENTITY,  name STRING COLLATE UTF8_LCASE,  category STRING COLLATE UTF8_LCASE,  price DECIMAL(10, 2));
-- Queries automatically case-insensitive (no LOWER() needed)SELECT * FROM catalog.schema.productsWHERE name = 'MacBook Pro';  -- matches 'macbook pro', 'MACBOOK PRO', etc.

-- Set up connection first (one-time)CREATE CONNECTION my_api_conn  TYPE HTTP  OPTIONS (host 'https://api.example.com', bearer_token secret('scope', 'token'));
-- Call API from SQLSELECT  order_id,  http_request(    conn => 'my_api_conn',    method => 'POST',    path => '/v1/validate',    json => to_json(named_struct('order_id', order_id, 'amount', amount))  ).text AS api_responseFROM catalog.schema.ordersWHERE needs_validation = true;

-- Read JSON files from a Volume with schema hintsSELECT *FROM read_files(  '/Volumes/catalog/schema/raw/events/',  format => 'json',  schemaHints => 'event_id STRING, timestamp TIMESTAMP, payload MAP<STRING, STRING>',  pathGlobFilter => '*.json',  recursiveFileLookup => true);
-- Read CSV with optionsSELECT *FROM read_files(  '/Volumes/catalog/schema/raw/sales/',  format => 'csv',  header => true,  delimiter => '|',  dateFormat => 'yyyy-MM-dd',  schema => 'sale_id INT, sale_date DATE, amount DECIMAL(10,2), store STRING');

WITH RECURSIVE org_chart AS (  -- Anchor: top-level managers  SELECT employee_id, name, manager_id, 0 AS depth, ARRAY(name) AS path  FROM catalog.schema.employees  WHERE manager_id IS NULL
  UNION ALL
  -- Recursive: direct reports  SELECT e.employee_id, e.name, e.manager_id, o.depth + 1, array_append(o.path, e.name)  FROM catalog.schema.employees e  JOIN org_chart o ON e.manager_id = o.employee_id  WHERE o.depth < 10  -- safety limit)SELECT * FROM org_chart ORDER BY depth, name;

-- Query PostgreSQL via Lakehouse FederationSELECT *FROM remote_query(  'my_postgres_connection',  database => 'my_database',  query    => 'SELECT customer_id, email, created_at FROM customers WHERE active = true');