-- Installeer SnowSQL: https://docs.snowflake.com/en/user-guide/snowsql-install-config

-- Verbind met Snowflake
!connect account=yz12345.eu-west-1
user=jansmit
password=************
warehouse=COMPUTE_WH
database=DEMO_DB
schema=PUBLIC
role=SYSADMIN;

-- Maak je eerste database
CREATE OR REPLACE DATABASE SALES_DB 
COMMENT = 'Productie database voor sales data';

-- Maak een nieuw schema
CREATE OR REPLACE SCHEMA SALES_DB.RAW 
COMMENT = 'Raw landing zone voor bron systemen';

-- Maak je eerste tabel
CREATE OR REPLACE TABLE SALES_DB.RAW.CUSTOMERS (
    CUSTOMER_ID NUMBER(38,0) AUTOINCREMENT START 1 INCREMENT 1,
    FIRST_NAME VARCHAR(100),
    LAST_NAME VARCHAR(100),
    EMAIL VARCHAR(255) UNIQUE,
    PHONE VARCHAR(20),
    ADDRESS VARCHAR(500),
    CITY VARCHAR(100),
    COUNTRY VARCHAR(100),
    CREATED_AT TIMESTAMP_NTZ(9) DEFAULT CURRENT_TIMESTAMP(),
    UPDATED_AT TIMESTAMP_NTZ(9) DEFAULT CURRENT_TIMESTAMP()
) 
COMMENT = 'Klant master data tabel';

-- Maak een stage voor data loading
CREATE OR REPLACE STAGE SALES_DB.RAW.CUSTOMER_STAGE
    URL = 's3://your-bucket/customer-data/'
    STORAGE_INTEGRATION = YOUR_S3_INTEGRATION
    FILE_FORMAT = (TYPE = 'CSV' SKIP_HEADER = 1);

-- Load data van S3
COPY INTO SALES_DB.RAW.CUSTOMERS
FROM @SALES_DB.RAW.CUSTOMER_STAGE
PATTERN = '.*customers.*\.csv'
ON_ERROR = 'CONTINUE';

-- Query data met Time Travel
SELECT * FROM SALES_DB.RAW.CUSTOMERS
AT(OFFSET => -60*5); -- Data van 5 minuten geleden

-- Toon alle databases
SHOW DATABASES;

-- Toon kosten van vandaag
SELECT 
    WAREHOUSE_NAME,
    SUM(CREDITS_USED) AS TOTAL_CREDITS,
    SUM(CREDITS_USED_COMPUTE) AS COMPUTE_CREDITS,
    SUM(CREDITS_USED_CLOUD_SERVICES) AS SERVICES_CREDITS
FROM INFORMATION_SCHEMA.WAREHOUSE_METERING_HISTORY
WHERE START_TIME >= CURRENT_DATE()
GROUP BY WAREHOUSE_NAME
ORDER BY TOTAL_CREDITS DESC;
        

# Installeer: pip install snowflake-connector-python snowflake-snowpark-python

from snowflake.snowpark import Session
from snowflake.snowpark.functions import col, sum as sum_, avg, count
import pandas as pd
import json

def create_snowflake_session():
    """Maak een Snowflake Snowpark sessie"""
    connection_params = {
        "account": "yz12345.eu-west-1",
        "user": "DATAPARTNER_USER",
        "password": "your_secure_password",
        "role": "DATA_ENGINEER",
        "warehouse": "COMPUTE_WH",
        "database": "ANALYTICS_DB",
        "schema": "PRODUCTION"
    }
    
    session = Session.builder.configs(connection_params).create()
    print("✅ Snowflake Snowpark sessie succesvol aangemaakt")
    return session

def demonstrate_snowpark_features(session):
    """Demonstreer Snowpark functionaliteiten"""
    
    # 1. Laad data in Snowpark DataFrame (lazy evaluation)
    sales_df = session.table("SALES_DB.RAW.SALES_TRANSACTIONS")
    
    print("Aantal rijen in sales:", sales_df.count())
    print("Schema van sales:", sales_df.schema)
    
    # 2. Data transformaties
    daily_sales = sales_df \
        .group_by("TRANSACTION_DATE") \
        .agg(
            sum_("AMOUNT").alias("TOTAL_SALES"),
            avg("AMOUNT").alias("AVG_TRANSACTION"),
            count("*").alias("TRANSACTION_COUNT")
        ) \
        .filter(col("TOTAL_SALES") > 10000) \
        .sort(col("TRANSACTION_DATE").desc())
    
    # 3. Toon SQL die gegenereerd wordt
    print("Generated SQL:")
    print(daily_sales.queries['queries'][0])
    
    # 4. Materialiseer de resultaten
    results = daily_sales.limit(10).collect()
    print("\nTop 10 verkoopdagen:")
    for row in results:
        print(f"{row.TRANSACTION_DATE}: €{row.TOTAL_SALES:,.2f}")
    
    # 5. Pandas integratie
    pandas_df = daily_sales.to_pandas()
    print(f"\nPandas DataFrame shape: {pandas_df.shape}")
    
    return pandas_df

def create_udf_for_calculations(session):
    """Maak een User Defined Function in Snowflake"""
    
    # Register een Python UDF
    @udf(name="calculate_discount", 
         input_types=[FloatType(), FloatType()], 
         return_type=FloatType(),
         is_permanent=True,
         stage_location="@PYTHON_UDFS",
         packages=["pandas"])
    def calculate_discount(price, discount_percentage):
        """Bereken de prijs na korting"""
        if discount_percentage > 50:
            discount_percentage = 50  # Max 50% korting
        return price * (1 - discount_percentage/100)
    
    print("✅ UDF succesvol geregistreerd in Snowflake")
    
    # Gebruik de UDF in een query
    result = session.sql("""
        SELECT 
            PRODUCT_ID,
            PRICE,
            DISCOUNT_PERCENTAGE,
            calculate_discount(PRICE, DISCOUNT_PERCENTAGE) as FINAL_PRICE
        FROM SALES_DB.RAW.PRODUCTS
        WHERE DISCOUNT_PERCENTAGE > 0
        LIMIT 10
    """).collect()
    
    return result

if __name__ == "__main__":
    # Maak Snowpark sessie
    session = create_snowflake_session()
    
    try:
        # Demo Snowpark features
        sales_data = demonstrate_snowpark_features(session)
        
        # Maak en gebruik UDF
        udf_results = create_udf_for_calculations(session)
        
        # Zero-copy cloning demo
        print("\n--- Zero-Copy Cloning Demo ---")
        session.sql("CREATE OR REPLACE DATABASE SALES_DB_CLONE CLONE SALES_DB").collect()
        print("Database succesvol gekloond zonder extra storage kosten!")
        
        # Time Travel demo
        print("\n--- Time Travel Demo ---")
        history = session.sql("""
            SELECT 
                MIN(CREATED) AS OLDEST_DATA,
                MAX(CREATED) AS NEWEST_DATA
            FROM SALES_DB.RAW.SALES_TRANSACTIONS
            AT(OFFSET => -3600) -- 1 uur geleden
        """).collect()
        print(f"Data van 1 uur geleden: {history[0]}")
        
    finally:
        session.close()
        print("\n✅ Sessie gesloten")
        

-- Warehouse voor de dagelijkse laadprocessen
CREATE WAREHOUSE laden_wh WITH
  WAREHOUSE_SIZE      = 'SMALL'
  AUTO_SUSPEND        = 60      -- na 60 seconden niksdoen vanzelf uit
  AUTO_RESUME         = TRUE    -- gaat weer aan zodra er een query komt
  INITIALLY_SUSPENDED = TRUE;   -- start uit, niet aan

-- Warehouse voor de dashboards, met ruimte voor gelijktijdige gebruikers
CREATE WAREHOUSE bi_wh WITH
  WAREHOUSE_SIZE      = 'MEDIUM'
  AUTO_SUSPEND        = 300
  AUTO_RESUME         = TRUE
  MIN_CLUSTER_COUNT   = 1
  MAX_CLUSTER_COUNT   = 3       -- meer clusters zodra de wachtrij oploopt
  SCALING_POLICY      = 'STANDARD';

-- Maat aanpassen kan terwijl hij draait; lopende queries gaan gewoon door
ALTER WAREHOUSE laden_wh SET WAREHOUSE_SIZE = 'MEDIUM';

-- Handmatig uitzetten
ALTER WAREHOUSE laden_wh SUSPEND;
        

-- ECONOMY wacht langer voor hij een cluster bijzet: goedkoper,
-- maar gebruikers staan soms even in de rij.
ALTER WAREHOUSE bi_wh SET
  MAX_CLUSTER_COUNT = 4
  SCALING_POLICY    = 'ECONOMY';

-- Kijken of er wachtrijen ontstaan in het afgelopen etmaal
SELECT
    warehouse_name,
    SUM(avg_running)      AS gemiddeld_actief,
    SUM(avg_queued_load)  AS gemiddeld_in_wachtrij
FROM snowflake.account_usage.warehouse_load_history
WHERE start_time >= DATEADD(day, -1, CURRENT_TIMESTAMP())
GROUP BY warehouse_name
ORDER BY gemiddeld_in_wachtrij DESC;
        

-- Alleen voor het meten: zet de resultaatcache uit in deze sessie
ALTER SESSION SET USE_CACHED_RESULT = FALSE;

-- Terugzetten als je klaar bent met meten
ALTER SESSION SET USE_CACHED_RESULT = TRUE;

-- Welke queries kwamen uit de cache?
SELECT
    query_text,
    warehouse_name,
    execution_status,
    total_elapsed_time / 1000 AS seconden,
    credits_used_cloud_services
FROM snowflake.account_usage.query_history
WHERE start_time >= DATEADD(hour, -2, CURRENT_TIMESTAMP())
  AND warehouse_name IS NULL   -- geen warehouse gebruikt = uit de cache
ORDER BY start_time DESC
LIMIT 20;
        

-- Vertel Snowflake hoe je bestanden eruitzien
CREATE OR REPLACE FILE FORMAT csv_formaat
  TYPE                         = 'CSV'
  FIELD_DELIMITER              = ';'
  SKIP_HEADER                  = 1
  FIELD_OPTIONALLY_ENCLOSED_BY = '"'
  NULL_IF                      = ('', 'NULL', 'null')
  EMPTY_FIELD_AS_NULL          = TRUE
  DATE_FORMAT                  = 'DD-MM-YYYY';

-- Externe stage op S3, met een storage integration voor de rechten
CREATE OR REPLACE STAGE orders_stage
  URL                 = 's3://mijn-bucket/orders/'
  STORAGE_INTEGRATION = s3_integratie
  FILE_FORMAT         = csv_formaat;

-- Kijken wat er klaarstaat zonder iets te laden
LIST @orders_stage;

-- En even in een bestand gluren
SELECT $1, $2, $3
FROM @orders_stage/orders_2026_01.csv
LIMIT 5;
        

CREATE OR REPLACE TABLE orders_bronze (
    order_id      NUMBER,
    klant_id      NUMBER,
    besteldatum   DATE,
    bedrag        NUMBER(12,2),
    status        VARCHAR,
    geladen_op    TIMESTAMP_NTZ DEFAULT CURRENT_TIMESTAMP()
);

-- Eerst droogzwemmen: valideren zonder te schrijven
COPY INTO orders_bronze
FROM @orders_stage
FILE_FORMAT = (FORMAT_NAME = csv_formaat)
VALIDATION_MODE = 'RETURN_ERRORS';

-- Daadwerkelijk laden
COPY INTO orders_bronze (order_id, klant_id, besteldatum, bedrag, status)
FROM (
    SELECT $1::NUMBER, $2::NUMBER, $3::DATE, $4::NUMBER(12,2), $5::VARCHAR
    FROM @orders_stage
)
FILE_FORMAT = (FORMAT_NAME = csv_formaat)
ON_ERROR     = 'CONTINUE';   -- foute rijen overslaan, de rest wel laden

-- Wat ging er mis bij de laatste lading?
SELECT * FROM TABLE(VALIDATE(orders_bronze, JOB_ID => '_last'));
        

CREATE OR REPLACE PIPE orders_pipe
  AUTO_INGEST = TRUE
AS
COPY INTO orders_bronze (order_id, klant_id, besteldatum, bedrag, status)
FROM (
    SELECT $1::NUMBER, $2::NUMBER, $3::DATE, $4::NUMBER(12,2), $5::VARCHAR
    FROM @orders_stage
)
FILE_FORMAT = (FORMAT_NAME = csv_formaat);

-- Dit ARN zet je als melding op je S3-bucket
SHOW PIPES;

-- Draait hij, en loopt er iets achter?
SELECT SYSTEM$PIPE_STATUS('orders_pipe');

-- Wat heeft de pipe de afgelopen dag gedaan?
SELECT file_name, status, row_count, first_error_message, last_load_time
FROM TABLE(snowflake.information_schema.copy_history(
    TABLE_NAME      => 'ORDERS_BRONZE',
    START_TIME      => DATEADD(day, -1, CURRENT_TIMESTAMP())
))
ORDER BY last_load_time DESC;
        

CREATE OR REPLACE TABLE events_raw (payload VARIANT);

COPY INTO events_raw
FROM @events_stage
FILE_FORMAT = (TYPE = 'JSON');

-- Velden eruit halen met puntnotatie; :: zet het om naar een echt type
SELECT
    payload:event_id::STRING            AS event_id,
    payload:gebruiker.naam::STRING      AS gebruikersnaam,
    payload:tijdstip::TIMESTAMP_NTZ     AS tijdstip,
    payload:bedrag::NUMBER(12,2)        AS bedrag
FROM events_raw
WHERE payload:type::STRING = 'bestelling';

-- Een array uitklappen naar losse rijen
SELECT
    e.payload:order_id::STRING AS order_id,
    r.value:sku::STRING        AS sku,
    r.value:aantal::NUMBER     AS aantal
FROM events_raw e,
     LATERAL FLATTEN(input => e.payload:regels) r;
        

-- Hoeveel partities raakten je zwaarste query's aan?
SELECT
    query_id,
    LEFT(query_text, 80)                          AS query,
    partitions_scanned,
    partitions_total,
    ROUND(100 * partitions_scanned / NULLIF(partitions_total, 0), 1) AS pct_gescand,
    bytes_spilled_to_local_storage                AS gemorst_lokaal,
    total_elapsed_time / 1000                     AS seconden
FROM snowflake.account_usage.query_history
WHERE start_time >= DATEADD(day, -7, CURRENT_TIMESTAMP())
  AND partitions_total > 0
ORDER BY partitions_scanned DESC
LIMIT 20;
        

-- Eerst kijken of het nodig is. Let op average_overlaps:
-- hoe hoger, hoe meer partities elkaar overlappen en hoe slechter het pruned.
SELECT SYSTEM$CLUSTERING_INFORMATION('orders', '(besteldatum, klant_id)');

-- Pas daarna instellen
ALTER TABLE orders CLUSTER BY (besteldatum, klant_id);

-- Wat kost het herordenen?
SELECT
    table_name,
    SUM(credits_used) AS credits,
    SUM(num_rows_reclustered) AS rijen_herordend
FROM snowflake.account_usage.automatic_clustering_history
WHERE start_time >= DATEADD(day, -30, CURRENT_TIMESTAMP())
GROUP BY table_name
ORDER BY credits DESC;

-- Weer uitzetten als het meer kost dan het oplevert
ALTER TABLE orders DROP CLUSTERING KEY;
        

-- FOUT: de functie om besteldatum heen maakt pruning onmogelijk
SELECT klant_id, SUM(bedrag)
FROM orders
WHERE YEAR(besteldatum) = 2026
GROUP BY klant_id;

-- GOED: een bereik op de kolom zelf, metadata blijft bruikbaar
SELECT klant_id, SUM(bedrag)
FROM orders
WHERE besteldatum >= '2026-01-01'
  AND besteldatum <  '2027-01-01'
GROUP BY klant_id;

-- FOUT: hoofdletterongevoelig zoeken sloopt het pruning ook
SELECT * FROM klanten WHERE UPPER(email) = 'INFO@VOORBEELD.NL';

-- GOED: sla de genormaliseerde waarde op bij het laden en filter daarop
SELECT * FROM klanten WHERE email_lower = 'info@voorbeeld.nl';
        

CREATE OR REPLACE MATERIALIZED VIEW omzet_per_dag AS
SELECT
    besteldatum,
    COUNT(*)      AS aantal_orders,
    SUM(bedrag)   AS omzet
FROM orders
GROUP BY besteldatum;

-- Wat kost het onderhoud ervan?
SELECT table_name, SUM(credits_used) AS credits
FROM snowflake.account_usage.materialized_view_refresh_history
WHERE start_time >= DATEADD(day, -30, CURRENT_TIMESTAMP())
GROUP BY table_name
ORDER BY credits DESC;
        

-- Drie manieren om terug te kijken
SELECT * FROM orders AT(OFFSET => -60 * 15);          -- 15 minuten geleden
SELECT * FROM orders AT(TIMESTAMP => '2026-10-24 08:00:00'::TIMESTAMP_NTZ);
SELECT * FROM orders BEFORE(STATEMENT => '01b2c3d4-...');  -- vlak voor die query

-- Wat is er precies veranderd sinds vanochtend?
SELECT COUNT(*) AS nu FROM orders;
SELECT COUNT(*) AS vanochtend FROM orders AT(TIMESTAMP => '2026-10-24 06:00:00'::TIMESTAMP_NTZ);

-- Per tabel instellen hoe ver je terug wilt kunnen
ALTER TABLE orders SET DATA_RETENTION_TIME_IN_DAYS = 30;

-- Een verwijderde tabel terughalen
UNDROP TABLE orders;
        

-- 1. Zoek de schuldige query op
SELECT query_id, query_text, user_name, start_time, rows_updated
FROM snowflake.account_usage.query_history
WHERE query_type = 'UPDATE'
  AND start_time >= DATEADD(hour, -4, CURRENT_TIMESTAMP())
ORDER BY start_time DESC;

-- 2. Controleer eerst of de oude stand er goed uitziet
SELECT * FROM orders BEFORE(STATEMENT => '01b2c3d4-0000-abcd-0000-000000000001')
LIMIT 20;

-- 3. Zet een kopie van de oude stand apart
CREATE TABLE orders_herstel AS
SELECT * FROM orders BEFORE(STATEMENT => '01b2c3d4-0000-abcd-0000-000000000001');

-- 4. Pas daarna omwisselen; SWAP is atomair
ALTER TABLE orders SWAP WITH orders_herstel;
        

-- Hele database klonen voor acceptatie
CREATE DATABASE acc_db CLONE prod_db;

-- Klonen zoals het er gisteren uitzag: Time Travel en cloning combineren
CREATE TABLE orders_gisteren
CLONE orders AT(TIMESTAMP => DATEADD(day, -1, CURRENT_TIMESTAMP()));

-- Veilig experimenteren met een zware wijziging
CREATE TABLE orders_test CLONE orders;
ALTER TABLE orders_test CLUSTER BY (klant_id);
-- meten, vergelijken, en daarna opruimen
DROP TABLE orders_test;
        

USE ROLE SECURITYADMIN;

CREATE ROLE analist;
CREATE ROLE engineer;

-- Analisten mogen lezen
GRANT USAGE   ON DATABASE prod_db            TO ROLE analist;
GRANT USAGE   ON SCHEMA   prod_db.gold       TO ROLE analist;
GRANT SELECT  ON ALL TABLES IN SCHEMA prod_db.gold TO ROLE analist;
-- ook op tabellen die er later bij komen
GRANT SELECT  ON FUTURE TABLES IN SCHEMA prod_db.gold TO ROLE analist;
GRANT USAGE   ON WAREHOUSE bi_wh             TO ROLE analist;

-- Engineers mogen alles wat analisten mogen, plus schrijven
GRANT ROLE analist TO ROLE engineer;
GRANT ALL ON SCHEMA prod_db.silver TO ROLE engineer;
GRANT USAGE ON WAREHOUSE laden_wh  TO ROLE engineer;

-- Pas daarna aan mensen toekennen
GRANT ROLE analist TO USER jansen;

-- Controleren wat een rol werkelijk mag
SHOW GRANTS TO ROLE analist;
        

-- E-mailadressen alleen zichtbaar voor wie dat nodig heeft
CREATE OR REPLACE MASKING POLICY email_masker AS (val STRING) RETURNS STRING ->
  CASE
    WHEN CURRENT_ROLE() IN ('ENGINEER', 'ACCOUNTADMIN') THEN val
    ELSE REGEXP_REPLACE(val, '.+@', '*****@')
  END;

ALTER TABLE klanten MODIFY COLUMN email
  SET MASKING POLICY email_masker;

-- Accountmanagers zien alleen hun eigen regio
CREATE OR REPLACE ROW ACCESS POLICY regio_filter AS (regio STRING) RETURNS BOOLEAN ->
  CURRENT_ROLE() = 'ENGINEER'
  OR regio = CURRENT_USER();   -- in de praktijk via een koppeltabel

ALTER TABLE klanten ADD ROW ACCESS POLICY regio_filter ON (regio);
        

USE ROLE ACCOUNTADMIN;

CREATE OR REPLACE RESOURCE MONITOR maandbudget WITH
  CREDIT_QUOTA = 500
  FREQUENCY    = MONTHLY
  START_TIMESTAMP = IMMEDIATELY
  TRIGGERS
    ON 75  PERCENT DO NOTIFY
    ON 90  PERCENT DO NOTIFY
    ON 100 PERCENT DO SUSPEND            -- lopende query's mogen afmaken
    ON 110 PERCENT DO SUSPEND_IMMEDIATE; -- hard stoppen

ALTER WAREHOUSE bi_wh SET RESOURCE_MONITOR = maandbudget;

-- Waar gaan de credits heen?
SELECT
    warehouse_name,
    ROUND(SUM(credits_used), 1) AS credits,
    ROUND(SUM(credits_used) * 2.50, 0) AS indicatie_euro  -- pas aan naar je eigen tarief
FROM snowflake.account_usage.warehouse_metering_history
WHERE start_time >= DATEADD(day, -30, CURRENT_TIMESTAMP())
GROUP BY warehouse_name
ORDER BY credits DESC;

-- Welke gebruikers verbruiken het meest?
SELECT
    user_name,
    COUNT(*)                                   AS aantal_queries,
    ROUND(SUM(total_elapsed_time) / 3600000, 1) AS draaiuren
FROM snowflake.account_usage.query_history
WHERE start_time >= DATEADD(day, -30, CURRENT_TIMESTAMP())
GROUP BY user_name
ORDER BY draaiuren DESC
LIMIT 15;