Snowflake Platform Digest

-- Step 1: Add a primary key with RELY to your upstream base or dynamic table
ALTER TABLE YOUR_DATABASE.YOUR_SCHEMA.ORDERS
  ADD PRIMARY KEY (order_id) RELY;

-- Step 2: Create downstream dynamic table with INCREMENTAL refresh
-- Snowflake will now use order_id to compute only changed rows
CREATE OR REPLACE DYNAMIC TABLE YOUR_DATABASE.YOUR_SCHEMA.ORDERS_SUMMARY
  TARGET_LAG = '5 minutes'
  WAREHOUSE = YOUR_WAREHOUSE
  REFRESH_MODE = INCREMENTAL
AS
SELECT
    customer_id,
    COUNT(*) AS order_count,
    SUM(amount) AS total_revenue
FROM YOUR_DATABASE.YOUR_SCHEMA.ORDERS
GROUP BY customer_id;

-- Step 3: Verify the refresh mode Snowflake actually selected
SHOW DYNAMIC TABLES LIKE 'ORDERS_SUMMARY' IN SCHEMA YOUR_DATABASE.YOUR_SCHEMA;
-- Check the "refresh_mode" and "refresh_mode_reason" columns

-- Call COMPLETE() with a system prompt to control output format and tone
SELECT
    review_id,
    review_text,
    SNOWFLAKE.CORTEX.COMPLETE(
        'mistral-large2',
        [
            {
                'role': 'system',
                'content': 'You are a customer sentiment analyst.
                Respond ONLY with valid JSON in this format:
                {"sentiment": "positive|neutral|negative",
                 "score": 0.0-1.0,
                 "key_theme": "one phrase"}.
                Never include explanation or markdown.'
            },
            {
                'role': 'user',
                'content': review_text
            }
        ],
        {
            'temperature': 0,
            'max_tokens': 100
        }
    )::VARIANT AS sentiment_analysis
FROM YOUR_DATABASE.YOUR_SCHEMA.CUSTOMER_REVIEWS
WHERE processed_flag = FALSE
LIMIT 1000;

-- Parse the JSON result downstream:
-- sentiment_analysis:sentiment::STRING, sentiment_analysis:score::FLOAT

-- Step 1: Create a resource monitor for your AI analytics warehouse
CREATE OR REPLACE RESOURCE MONITOR ai_workload_monitor
  CREDIT_QUOTA = 500
  FREQUENCY = MONTHLY
  START_TIMESTAMP = IMMEDIATELY
  TRIGGERS
    ON 75 PERCENT DO NOTIFY
    ON 90 PERCENT DO NOTIFY
    ON 100 PERCENT DO SUSPEND;

-- Step 2: Attach monitor to your AI-specific warehouse
ALTER WAREHOUSE YOUR_AI_WAREHOUSE
  SET RESOURCE_MONITOR = ai_workload_monitor;

-- Step 3: Query current credit consumption
SELECT
    name,
    credit_quota,
    credits_used,
    ROUND(credits_used / credit_quota * 100, 1) AS pct_consumed,
    remaining_credits
FROM snowflake.account_usage.resource_monitors
WHERE name = 'AI_WORKLOAD_MONITOR';

-- Step 4: See which Cortex functions consumed the most credits this month
SELECT
    function_name,
    COUNT(*) AS call_count,
    SUM(credits_used) AS total_credits
FROM snowflake.account_usage.metering_history
WHERE service_type = 'AI_SERVICES'
  AND start_time >= DATE_TRUNC('month', CURRENT_DATE)
GROUP BY function_name
ORDER BY total_credits DESC;

-- Step 1: Stage your PDFs in an internal or external stage
-- PUT file:///local/path/invoice.pdf @YOUR_DATABASE.YOUR_SCHEMA.DOC_STAGE;

-- Step 2: Parse raw text and layout from the document
CREATE OR REPLACE TABLE YOUR_DATABASE.YOUR_SCHEMA.PARSED_INVOICES AS
SELECT
    relative_path AS file_name,
    SNOWFLAKE.CORTEX.AI_PARSE_DOCUMENT(
        @YOUR_DATABASE.YOUR_SCHEMA.DOC_STAGE,
        relative_path,
        {'mode': 'LAYOUT'}
    ) AS parsed_doc
FROM directory(@YOUR_DATABASE.YOUR_SCHEMA.DOC_STAGE)
WHERE relative_path ILIKE '%.pdf';

-- Step 3: Extract structured fields using AI_EXTRACT with a JSON schema
SELECT
    file_name,
    SNOWFLAKE.CORTEX.AI_EXTRACT(
        parsed_doc,
        {
            'invoice_number': {'type': 'string',  'description': 'Invoice ID or number'},
            'vendor_name':    {'type': 'string',  'description': 'Vendor or supplier name'},
            'invoice_date':   {'type': 'string',  'description': 'Date of invoice YYYY-MM-DD'},
            'total_amount':   {'type': 'number',  'description': 'Total invoice amount (numeric)'},
            'currency':       {'type': 'string',  'description': 'Currency code (USD, EUR, etc.)'}
        }
    )::VARIANT AS extracted_fields
FROM YOUR_DATABASE.YOUR_SCHEMA.PARSED_INVOICES;

-- Step 4: Flatten and use the results as structured data
-- extracted_fields:invoice_number::STRING
-- extracted_fields:total_amount::FLOAT

-- Step 1: Grant base Cortex access to all data users
GRANT DATABASE ROLE SNOWFLAKE.CORTEX_USER TO ROLE DATA_ANALYST;

-- Step 2: Grant AI function-specific access only to approved roles
GRANT DATABASE ROLE SNOWFLAKE.AI_FUNCTIONS_USER TO ROLE AI_DEVELOPER;
GRANT DATABASE ROLE SNOWFLAKE.AI_FUNCTIONS_USER TO ROLE DATA_SCIENTIST;

-- Step 3: Verify which roles have access to AI functions
SHOW GRANTS OF DATABASE ROLE SNOWFLAKE.AI_FUNCTIONS_USER;

-- Step 4: Review which users are calling AI functions
SELECT
    user_name,
    query_text,
    start_time,
    execution_status,
    ROUND(total_elapsed_time / 1000, 2) AS elapsed_seconds
FROM snowflake.account_usage.query_history
WHERE start_time >= DATEADD(day, -7, CURRENT_TIMESTAMP())
  AND (
    query_text ILIKE '%CORTEX.COMPLETE%'
    OR query_text ILIKE '%AI_EXTRACT%'
    OR query_text ILIKE '%AI_PARSE_DOCUMENT%'
    OR query_text ILIKE '%CORTEX.FINETUNE%'
  )
ORDER BY start_time DESC;

-- Create a dynamic Iceberg table with optimized partitioning
-- GA as of April 13, 2026 — supports PARTITION BY, TARGET_FILE_SIZE, PATH_LAYOUT
CREATE OR REPLACE DYNAMIC ICEBERG TABLE
    YOUR_DATABASE.YOUR_SCHEMA.EVENTS_DAILY_ICEBERG
  TARGET_LAG        = '1 hour'
  WAREHOUSE         = YOUR_WAREHOUSE
  EXTERNAL_VOLUME   = YOUR_EXTERNAL_VOLUME
  CATALOG           = SNOWFLAKE
  BASE_LOCATION     = 'events/daily/'
  PARTITION BY      (event_date, region)
  TARGET_FILE_SIZE  = 128MB
  PATH_LAYOUT       = HIVE
  REFRESH_MODE      = AUTO
AS
SELECT
    DATE_TRUNC('day', event_timestamp) AS event_date,
    region,
    event_type,
    user_id,
    COUNT(*) AS event_count,
    SUM(revenue) AS total_revenue
FROM YOUR_DATABASE.YOUR_SCHEMA.RAW_EVENTS
GROUP BY 1, 2, 3, 4;

-- Check the physical layout and file statistics
SELECT * FROM TABLE(
    INFORMATION_SCHEMA.DYNAMIC_TABLE_REFRESH_HISTORY(
        TABLE_NAME => 'EVENTS_DAILY_ICEBERG'
    )
)
ORDER BY refresh_start_time DESC
LIMIT 10;