BigQuery Stored Procedure Example

bq_stored_proc.sql

CREATE OR REPLACE PROCEDURE ledger.make_ledger_entries() BEGIN
  
  DECLARE unique_item_names ARRAY<STRING>;
  DECLARE item_idx INT64 DEFAULT 0;
  DECLARE current_table_name STRING;
  
  -- Create some fake staging data for illustration
  -- In reality, this could be ingested from a federated data source
  CREATE OR REPLACE TEMPORARY TABLE new_purchases AS (
    SELECT *
    FROM UNNEST([
      STRUCT("apple" AS item_name, 2 AS quantity, TIMESTAMP '2020-10-01T00:00:00' AS purchased_at),
            ("banana",             10,            TIMESTAMP '2020-10-01T01:00:00'),
            ("pear",               4,             TIMESTAMP '2020-10-01T00:02:00'),
            ("apple",              4,             TIMESTAMP '2020-11-01T00:00:00')
    ])
  );
  
  -- Find distinct items in staging data
  SET unique_item_names = (SELECT ARRAY_AGG(DISTINCT item_name) FROM new_purchases);
  
  -- Insert entries from each item group into their own persistent tables
  WHILE item_idx < ARRAY_LENGTH(unique_item_names) DO
    
    SET current_table_name = CONCAT(unique_item_names[OFFSET(item_idx)], '_ledger');
    
    EXECUTE IMMEDIATE format("""
    CREATE TABLE IF NOT EXISTS ledger.%s AS 
    SELECT quantity, purchased_at 
      FROM new_purchases 
     WHERE false""", current_table_name);
     
    EXECUTE IMMEDIATE format("""
    INSERT ledger.%s (quantity, purchased_at) 
    SELECT quantity, purchased_at 
      FROM new_purchases
     WHERE item_name = '%s'""", current_table_name, unique_item_names[OFFSET(item_idx)]);
      
    SET item_idx = item_idx + 1;
    
  END WHILE;
  
END;

-- Lite it up
BEGIN
  CALL ledger.make_ledger_entries();
EXCEPTION WHEN ERROR THEN
  SELECT
    @@error.message,
    @@error.stack_trace,
    @@error.statement_text,
    @@error.formatted_stack_trace;
END;