Overview

---

The Problem

Iceberg everywhere + Snowpark Connect as the unified compute layer + Horizon for governance + Cortex for AI

---

What You Build

Download the code: Download all files (ZIP)

Or download individual files:

• 01_sf_iceberg_catalog_setup.sql
• 02_sf_iceberg_demo.ipynb
• 03_databricks_rw_sf_iceberg.py
• 04_databricks_create_uc_tables.py
• 05_databricks_federation_demo.ipynb
• 06_cortex_ai_pipeline.ipynb
• 07_ai_pipeline.ipynb

What You'll Learn

Key Capabilities

What You'll Need

Prerequisites

Architecture

Scenario 1 — Snowflake-Managed Iceberg + Snowpark Connect

Snowflake Account
  │  CREATE ICEBERG TABLE ... CATALOG = 'SNOWFLAKE'
  │  Horizon governance: column masking, row access policy
  │
  └─ Snowpark Connect
       spark.table("DEMO_SCHEMA.OPEN_TABLE")
       spark.table("DEMO_SCHEMA.PROTECTED_TABLE")
       → queries route through Snowflake SQL engine
       → Horizon policies ARE enforced

Scenario 2 — Federate External Iceberg + Cortex AI Enrichment

External Catalog (Unity Catalog)
  │  Delta + Iceberg UniForm tables
  │  Iceberg REST endpoint → Snowflake Catalog Integration
  │
Snowflake Catalog-Linked Database (DATABRICKS_DEMO_DB)
  │  Auto-discovers schemas + tables every 30 s
  │  Applies independent Horizon masking policies
  │
  └─ Snowpark Connect
       spark.sql("SELECT * FROM DATABRICKS_DEMO_DB.horizon_demo.customer_orders")
       sf_session.sql(CTAS + CORTEX.COMPLETE → AI_ORDER_INSIGHTS)
       → Snowflake SQL engine enforces masking on both raw and AI columns
       → Cortex Analyst: natural language queries on AI-enriched Iceberg table

Governance contrast (external engine vs Snowpark Connect): An external Spark engine reading via Horizon IRC receives raw Parquet via vended S3 credentials — Snowflake SQL-layer policies do not apply to that path. Snowpark Connect routes through the Snowflake SQL engine, so Horizon masking and row access policies are always enforced. See 03_databricks_rw_sf_iceberg.py for the full Databricks IRC demo.

Databricks Cluster Setup (Scenario 2 prerequisites)

Scenario 1 — Snowflake Setup

Create Database and Tables

USE ROLE ACCOUNTADMIN;
USE WAREHOUSE LOAD_WH;

CREATE DATABASE IF NOT EXISTS HORIZON_DEMO_SFDB
    EXTERNAL_VOLUME = 'SNOWFLAKE_MANAGED'
    COMMENT = 'Snowflake-managed Iceberg catalog (Horizon IRC)';

CREATE SCHEMA IF NOT EXISTS HORIZON_DEMO_SFDB.DEMO_SCHEMA;

-- OPEN_TABLE: no governance restrictions
CREATE OR REPLACE ICEBERG TABLE HORIZON_DEMO_SFDB.DEMO_SCHEMA.OPEN_TABLE (
    id          INT,
    product     STRING,
    quantity    INT,
    price       DECIMAL(10, 2),
    created_at  TIMESTAMP
)
CATALOG = 'SNOWFLAKE';

-- PROTECTED_TABLE: column masking + row access policy applied below
CREATE OR REPLACE ICEBERG TABLE HORIZON_DEMO_SFDB.DEMO_SCHEMA.PROTECTED_TABLE (
    id              INT,
    customer_name   STRING,
    sensitive_data  STRING,
    amount          DECIMAL(10, 2),
    created_at      TIMESTAMP
)
CATALOG = 'SNOWFLAKE';

Apply Horizon Governance Policies

-- Column masking: hide sensitive_data from non-admin roles
CREATE OR REPLACE MASKING POLICY HORIZON_DEMO_SFDB.DEMO_SCHEMA.MASK_SENSITIVE
    AS (val STRING) RETURNS STRING ->
    CASE
        WHEN CURRENT_ROLE() IN ('ACCOUNTADMIN', 'SYSADMIN') THEN val
        ELSE '*** MASKED ***'
    END;

ALTER ICEBERG TABLE HORIZON_DEMO_SFDB.DEMO_SCHEMA.PROTECTED_TABLE
    MODIFY COLUMN sensitive_data
    SET MASKING POLICY HORIZON_DEMO_SFDB.DEMO_SCHEMA.MASK_SENSITIVE;

-- Row access policy: non-admin roles see current-year rows only
CREATE OR REPLACE ROW ACCESS POLICY HORIZON_DEMO_SFDB.DEMO_SCHEMA.CURRENT_YEAR_ONLY
    AS (created_at TIMESTAMP) RETURNS BOOLEAN ->
        CURRENT_ROLE() IN ('ACCOUNTADMIN', 'SYSADMIN')
        OR YEAR(created_at) = YEAR(CURRENT_DATE());

ALTER ICEBERG TABLE HORIZON_DEMO_SFDB.DEMO_SCHEMA.PROTECTED_TABLE
    ADD ROW ACCESS POLICY HORIZON_DEMO_SFDB.DEMO_SCHEMA.CURRENT_YEAR_ONLY
    ON (created_at);

Configure Credential Vending (for Scenario 2)

-- OPEN_TABLE_RW: SELECT + writes → Snowflake vends write-capable S3 credentials
CREATE DATABASE ROLE IF NOT EXISTS HORIZON_DEMO_SFDB.OPEN_TABLE_RW;
GRANT SELECT, INSERT, UPDATE, DELETE ON TABLE
    HORIZON_DEMO_SFDB.DEMO_SCHEMA.OPEN_TABLE
    TO DATABASE ROLE HORIZON_DEMO_SFDB.OPEN_TABLE_RW;

-- PROTECTED_TABLE_RO: SELECT only → Snowflake vends read-only S3 credentials
CREATE DATABASE ROLE IF NOT EXISTS HORIZON_DEMO_SFDB.PROTECTED_TABLE_RO;
GRANT SELECT ON TABLE
    HORIZON_DEMO_SFDB.DEMO_SCHEMA.PROTECTED_TABLE
    TO DATABASE ROLE HORIZON_DEMO_SFDB.PROTECTED_TABLE_RO;

GRANT DATABASE ROLE HORIZON_DEMO_SFDB.OPEN_TABLE_RW      TO ROLE $SF_EXT_COMPUTE_ROLE;
GRANT DATABASE ROLE HORIZON_DEMO_SFDB.PROTECTED_TABLE_RO TO ROLE $SF_EXT_COMPUTE_ROLE;

Generate PAT (for Scenario 2)

-- ⚠ Copy the token immediately — shown only once
ALTER USER <YOUR_USERNAME>
    ADD PROGRAMMATIC ACCESS TOKEN MY_DEMO_PAT
    COMMENT = 'Iceberg Federation Demo — Horizon IRC integration';

Scenario 1 — Snowpark Connect Governance Demo

snowpark-connect is pre-installed in Snowflake Workspaces — no package picker step needed.

Initialize Session

from snowflake.snowpark.context import get_active_session
from snowflake import snowpark_connect
from pyspark.sql.functions import col
import pandas as pd

sf_session = get_active_session()
sf_session.sql(f"USE DATABASE {SF_MANAGED_ICEBERG_DB}").collect()

spark = snowpark_connect.init_spark_session()

def switch_role(role):
    sf_session.sql(f"USE ROLE {role}").collect()
    print(f"Active role → {role}")

switch_role("ACCOUNTADMIN")

Demo 1 — Read OPEN_TABLE

spark.table(TBL_OPEN).orderBy("id").show(truncate=False)

Demo 2 — ACCOUNTADMIN Reads PROTECTED_TABLE

switch_role("ACCOUNTADMIN")
spark.table(TBL_PROTECTED).orderBy("id").show(truncate=False)

Demo 3 — Reader Role Reads PROTECTED_TABLE

switch_role(SF_READER_ROLE)
spark.table(TBL_PROTECTED).orderBy("id").show(truncate=False)

Demo 4 — Side-by-Side Comparison

COLS = ["id", "customer_name", "sensitive_data", "amount"]

switch_role("ACCOUNTADMIN")
admin_rows = spark.table(TBL_PROTECTED).select(*COLS).collect()

switch_role(SF_READER_ROLE)
reader_rows = spark.table(TBL_PROTECTED).select(*COLS).collect()

admin_pd  = pd.DataFrame([r.asDict() for r in admin_rows])
admin_pd.insert(0, "role", "ACCOUNTADMIN")

reader_pd = pd.DataFrame([r.asDict() for r in reader_rows])
reader_pd.insert(0, "role", SF_READER_ROLE)

combined = pd.concat([admin_pd, reader_pd]).sort_values(["id", "role"]).reset_index(drop=True)
print(combined.to_string(index=False))

Key insight: Snowpark Connect routes queries through the Snowflake SQL engine. Horizon policies apply to every query regardless of the PySpark API used. Scenario 2 demonstrates the contrast — an external engine reading the same table bypasses these policies.

Governance Contrast — Snowpark Connect vs External Engine

switch_role("EXT_COMPUTE_ENG_DEMO_ROLE")
spark.table("HORIZON_DEMO_SFDB.DEMO_SCHEMA.PROTECTED_TABLE").show()
# 2 rows (2023 row filtered) │ sensitive_data: *** MASKED ***

# Databricks reads via vended S3 credentials
spark.table("sf_horizon.DEMO_SCHEMA.PROTECTED_TABLE").show()
# 3 rows (all rows) │ sensitive_data: raw value visible

Credential vending also enforces write access: PROTECTED_TABLE_RO (SELECT only) → Snowflake vends read-only S3 credentials → any s3:PutObject attempt returns 403. OPEN_TABLE_RW (INSERT/UPDATE/DELETE) → write-capable credentials vended. No application code required — the catalog decides.

Scenario 2 — External Catalog Setup

Create Catalog, Schema, and Tables

CATALOG_NAME = "<DBX_UC_CATALOG>"    # e.g. my_demo
SCHEMA_NAME  = "horizon_demo"         # pre-set

spark.sql(f"CREATE CATALOG IF NOT EXISTS {CATALOG_NAME}")
spark.sql(f"CREATE SCHEMA  IF NOT EXISTS {CATALOG_NAME}.{SCHEMA_NAME}")

spark.sql(f"""
CREATE OR REPLACE TABLE {CATALOG_NAME}.{SCHEMA_NAME}.customer_orders (
    order_id BIGINT, customer_id BIGINT, product STRING,
    amount DECIMAL(10,2), order_date DATE, status STRING
)
USING DELTA
TBLPROPERTIES (
    'delta.universalFormat.enabledFormats' = 'iceberg',
    'delta.enableIcebergCompatV2'          = 'true',
    'delta.columnMapping.mode'             = 'name'
)
""")

Apply External Catalog Governance

# Column mask on credit_card (applies within the external catalog only)
spark.sql(f"""
CREATE OR REPLACE FUNCTION {CATALOG_NAME}.{SCHEMA_NAME}.mask_credit_card(cc STRING)
RETURN CASE WHEN is_account_group_member('account_unity_admin') THEN cc
            ELSE CONCAT('****-****-****-', RIGHT(cc, 4)) END
""")
spark.sql(f"""ALTER TABLE {CATALOG_NAME}.{SCHEMA_NAME}.sensitive_orders
ALTER COLUMN credit_card SET MASK {CATALOG_NAME}.{SCHEMA_NAME}.mask_credit_card""")

These external catalog policies apply within that catalog only. Snowflake applies its own independent governance policies in the next step.

Get the Iceberg REST Endpoint

WORKSPACE_HOST = "<DBX_WORKSPACE_HOST>"
print(f"CATALOG_URI  : https://{WORKSPACE_HOST}/api/2.1/unity-catalog/iceberg-rest")
print(f"CATALOG_NAME : {CATALOG_NAME}")

Scenario 2 — Snowflake Setup

Create Catalog Integration

CREATE OR REPLACE CATALOG INTEGRATION MY_DATABRICKS_UC_CI
    CATALOG_SOURCE = ICEBERG_REST
    TABLE_FORMAT   = ICEBERG
    REST_CONFIG = (
        CATALOG_URI  = 'https://<DBX_WORKSPACE_HOST>/api/2.1/unity-catalog/iceberg-rest'
        CATALOG_NAME = '<DBX_UC_CATALOG>'
        ACCESS_DELEGATION_MODE = EXTERNAL_VOLUME_CREDENTIALS
    )
    REST_AUTHENTICATION = (
        TYPE         = BEARER
        BEARER_TOKEN = '<DBX_PAT_TOKEN>'
    )
    ENABLED = TRUE;

-- Verify connectivity — must return {"success": true}
SELECT SYSTEM$VERIFY_CATALOG_INTEGRATION('MY_DATABRICKS_UC_CI');

Create Catalog-Linked Database

CREATE DATABASE IF NOT EXISTS DATABRICKS_DEMO_DB
    EXTERNAL_VOLUME = 'ICEBERG_EXTERNAL_S3_VOLUME'
    LINKED_CATALOG = ( CATALOG = 'MY_DATABRICKS_UC_CI' )
    COMMENT = 'Iceberg Federation Demo — Federated from external Iceberg catalog';

-- Wait ~30 s for auto-discovery, then verify
SHOW ICEBERG TABLES IN DATABASE DATABRICKS_DEMO_DB;
SELECT * FROM DATABRICKS_DEMO_DB.horizon_demo.customer_orders LIMIT 5;

Apply Snowflake Horizon Governance

CREATE OR REPLACE MASKING POLICY HORIZON_DEMO_SFDB.GOVERNANCE_POLICIES.MASK_CREDIT_CARD
    AS (val STRING) RETURNS STRING ->
    CASE
        WHEN CURRENT_ROLE() = 'ACCOUNTADMIN' THEN val
        ELSE CONCAT('****-****-****-', RIGHT(val, 4))
    END;

ALTER ICEBERG TABLE DATABRICKS_DEMO_DB.horizon_demo.sensitive_orders
    MODIFY COLUMN "credit_card"
    SET MASKING POLICY HORIZON_DEMO_SFDB.GOVERNANCE_POLICIES.MASK_CREDIT_CARD;

Scenario 2 — Snowpark Connect Governance Demo

snowpark-connect is pre-installed in Snowflake Workspaces — no package picker step needed.

Session Setup for Catalog-Linked Databases

# Rule 1: session context must be a regular (non-catalog-linked) database
sf_session.sql(f"USE DATABASE {SF_INIT_DB}").collect()

# Rule 2: preserve lowercase identifiers used by the external catalog
conf = SparkConf().set("spark.sql.caseSensitive", "true")
spark = snowpark_connect.init_spark_session(conf=conf)

# Rule 3: fully-qualified reference with lowercase schema and table names
df = spark.sql(f"SELECT * FROM {SF_FEDERATED_DB}.{DBX_SCHEMA}.customer_orders")

Initialize Sessions

from snowflake.snowpark.context import get_active_session
from snowflake import snowpark_connect
from pyspark import SparkConf
from pyspark.sql.functions import col, count, sum as spark_sum

sf_session = get_active_session()
sf_session.sql(f"USE DATABASE {SF_INIT_DB}").collect()

conf = SparkConf().set("spark.sql.caseSensitive", "true")
spark = snowpark_connect.init_spark_session(conf=conf)

def switch_role(role):
    sf_session.sql(f"USE ROLE {role}").collect()
    print(f"Active role → {role}")

switch_role("ACCOUNTADMIN")

Demo 1 — Read customer_orders

df_orders = spark.sql(f"SELECT * FROM {TBL_ORDERS}")
df_orders.orderBy("order_id").show(truncate=False)

Demo 2–3 — Filter and Aggregate

df_orders.filter(col("status").isin("SHIPPED", "DELIVERED")) \
    .select("order_id", "product", "amount", "status") \
    .orderBy("amount", ascending=False).show(truncate=False)

df_orders.groupBy("status") \
    .agg(count("*").alias("order_count"), spark_sum("amount").alias("total_revenue")) \
    .orderBy("total_revenue", ascending=False).show(truncate=False)

Demo 4–5 — Role-Based Masking on Federated Tables

# ACCOUNTADMIN: credit_card unmasked
switch_role("ACCOUNTADMIN")
spark.sql(f"SELECT * FROM {TBL_SENSITIVE}") \
    .select("order_id", "customer_name", "credit_card").show(truncate=False)

# Reader role: credit_card masked by Snowflake Horizon policy
switch_role(SF_READER_ROLE)
spark.sql(f"SELECT * FROM {TBL_SENSITIVE}") \
    .select("order_id", "customer_name", "credit_card").show(truncate=False)

Demo 6 — Side-by-Side Governance

import pandas as pd
COLS = ["order_id", "customer_name", "credit_card"]

switch_role("ACCOUNTADMIN")
admin_rows  = spark.sql(f"SELECT * FROM {TBL_SENSITIVE}").select(*COLS).collect()

switch_role(SF_READER_ROLE)
reader_rows = spark.sql(f"SELECT * FROM {TBL_SENSITIVE}").select(*COLS).collect()

admin_pd  = pd.DataFrame([r.asDict() for r in admin_rows])
admin_pd.insert(0, "role", "ACCOUNTADMIN")
reader_pd = pd.DataFrame([r.asDict() for r in reader_rows])
reader_pd.insert(0, "role", SF_READER_ROLE)

combined = pd.concat([admin_pd, reader_pd]).sort_values(["order_id","role"]).reset_index(drop=True)
print(combined.to_string(index=False))

Scenario 2 — AI Enrichment Pipeline Setup

Step 1 — Verify Cortex Is Available

SELECT SNOWFLAKE.CORTEX.COMPLETE('llama3.1-8b', 'Reply with exactly one word: ready') AS cortex_status;

Trial availability: SNOWFLAKE.CORTEX.COMPLETE and SNOWFLAKE.CORTEX.CLASSIFY_TEXT are available in standard 30-day trial accounts on AWS us-east-1, us-west-2, and Azure eastus2. Cortex Analyst semantic views are available on the same regions.

Step 2 — Create AI_ORDER_INSIGHTS Iceberg Table

CREATE OR REPLACE ICEBERG TABLE HORIZON_DEMO_SFDB.DEMO_SCHEMA.AI_ORDER_INSIGHTS
    CATALOG = 'SNOWFLAKE'
    AS
WITH deduped_orders AS (
    SELECT order_id, customer_id, product, amount, order_date, status
    FROM DATABRICKS_DEMO_DB.horizon_demo.customer_orders
    QUALIFY ROW_NUMBER() OVER (PARTITION BY order_id ORDER BY order_id) = 1
),
deduped_sensitive AS (
    SELECT order_id, region
    FROM DATABRICKS_DEMO_DB.horizon_demo.sensitive_orders
    QUALIFY ROW_NUMBER() OVER (PARTITION BY order_id ORDER BY order_id) = 1
)
SELECT
    co.order_id, co.customer_id, co.product, co.amount, co.order_date, co.status,
    COALESCE(so.region, 'UNKNOWN') AS region,
    -- Deterministic risk tier (reliable for governance demo)
    CASE
        WHEN co.status = 'CANCELLED' THEN 'HIGH'
        WHEN co.amount >= 500        THEN 'HIGH'
        WHEN co.amount >= 100        THEN 'MEDIUM'
        ELSE 'LOW'
    END AS risk_level,
    -- Cortex-generated operational note
    SNOWFLAKE.CORTEX.COMPLETE(
        'llama3.1-8b',
        'Write one operational sentence (max 15 words) for logistics. Product: ' || co.product ||
        ', Amount: $' || co.amount::STRING || ', Status: ' || co.status || '.'
    )::VARCHAR AS ops_note,
    CURRENT_TIMESTAMP()::TIMESTAMP_LTZ(6) AS enriched_at
FROM deduped_orders co
LEFT JOIN deduped_sensitive so ON co.order_id = so.order_id;

Why QUALIFY ROW_NUMBER(): The source federated tables may contain duplicate rows from multiple Databricks INSERT runs. QUALIFY deduplicates per order_id before enrichment.

Why deterministic CASE WHEN for risk_level: LLMs produce inconsistent results for strict classification rules. Using CASE WHEN ensures reliable HIGH/MEDIUM/LOW splits for the governance demo. CORTEX.COMPLETE is used for ops_note where natural language generation excels.

Step 3 — Re-Apply Masking Policy

-- CREATE OR REPLACE drops all column policies — must re-apply every time
ALTER ICEBERG TABLE HORIZON_DEMO_SFDB.DEMO_SCHEMA.AI_ORDER_INSIGHTS
    MODIFY COLUMN risk_level
    SET MASKING POLICY HORIZON_DEMO_SFDB.DEMO_SCHEMA.MASK_RISK_LEVEL;

Important: CREATE OR REPLACE ICEBERG TABLE drops all column policies on the table. Always run this ALTER ICEBERG TABLE immediately after the CTAS. The 06_cortex_ai_pipeline.ipynb script does this automatically.

Step 4 — Governance Comparison on AI Output

-- ACCOUNTADMIN: sees actual risk level including HIGH
USE ROLE ACCOUNTADMIN;
SELECT 'ACCOUNTADMIN' AS role_context, order_id, product, amount, region, risk_level
FROM HORIZON_DEMO_SFDB.DEMO_SCHEMA.AI_ORDER_INSIGHTS ORDER BY order_id;

-- Reader role: HIGH masked to *** RESTRICTED ***
USE ROLE EXT_COMPUTE_ENG_DEMO_ROLE;
SELECT 'EXT_COMPUTE_ENG_DEMO_ROLE' AS role_context, order_id, product, amount, region, risk_level
FROM HORIZON_DEMO_SFDB.DEMO_SCHEMA.AI_ORDER_INSIGHTS ORDER BY order_id;

Step 5 — Create Cortex Analyst Semantic View

CREATE OR REPLACE SEMANTIC VIEW HORIZON_DEMO_SFDB.DEMO_SCHEMA.ICEBERG_AI_SEMANTIC_VIEW
TABLES (
  orders     AS HORIZON_DEMO_SFDB.DEMO_SCHEMA.AI_ORDER_INSIGHTS
               PRIMARY KEY (order_id)
               WITH SYNONYMS = ('ai orders', 'enriched orders', 'risk orders'),
  fed_orders AS DATABRICKS_DEMO_DB.horizon_demo.customer_orders
               PRIMARY KEY (order_id)
               WITH SYNONYMS = ('federated orders', 'source orders')
)
RELATIONSHIPS (
  orders(order_id) REFERENCES fed_orders
)
FACTS (
  orders.amount AS amount WITH SYNONYMS = ('order value', 'revenue')
)
DIMENSIONS (
  orders.product    AS product,
  orders.status     AS status     WITH SYNONYMS = ('fulfillment status'),
  orders.region     AS region,
  orders.risk_level AS risk_level WITH SYNONYMS = ('risk', 'risk tier'),
  orders.order_date AS order_date,
  orders.ops_note   AS ops_note   WITH SYNONYMS = ('operational note', 'ai note')
)
METRICS (
  orders.total_revenue   AS SUM(orders.amount)     WITH SYNONYMS = ('revenue', 'total sales'),
  orders.order_count     AS COUNT(orders.order_id) WITH SYNONYMS = ('number of orders'),
  orders.avg_order_value AS AVG(orders.amount)      WITH SYNONYMS = ('average order', 'AOV')
);

GRANT SELECT ON SEMANTIC VIEW HORIZON_DEMO_SFDB.DEMO_SCHEMA.ICEBERG_AI_SEMANTIC_VIEW
    TO ROLE EXT_COMPUTE_ENG_DEMO_ROLE;

Semantic views are the recommended path for Cortex Analyst — they replace legacy YAML files and appear automatically in Snowsight → AI & ML → Cortex Analyst. Note: the alias direction in TABLES is alias AS db.schema.table (not table AS alias).

Scenario 2 — Snowpark Connect AI Pipeline Notebook

snowpark-connect is pre-installed in Snowflake Workspaces — no package picker step needed.

Initialize Session

sf_session = get_active_session()
sf_session.sql(f"USE DATABASE {SF_INIT_DB}").collect()

conf = SparkConf().set("spark.sql.caseSensitive", "true")
spark = snowpark_connect.init_spark_session(conf=conf)

caseSensitive=true is required when reading catalog-linked (FDN) tables, which use lowercase identifiers. It is harmless when reading SF-managed tables.

Step 2 — Snowpark Connect Reads Federated Source Tables

# Read from CLD — federated Databricks Iceberg (read only)
df_orders = spark.sql(f"SELECT * FROM {TBL_ORDERS}")
df_orders.select("order_id", "product", "amount", "status").orderBy("order_id").show()

Step 4 — Cortex Enrichment + Write via sf_session

sf_session.sql(f"""
    CREATE OR REPLACE ICEBERG TABLE {TBL_INSIGHTS} CATALOG = 'SNOWFLAKE' AS
    WITH deduped_orders AS ( ... ),
         deduped_sensitive AS ( ... )
    SELECT ..., SNOWFLAKE.CORTEX.COMPLETE('llama3.1-8b', ...) AS ops_note ...
    FROM deduped_orders co LEFT JOIN deduped_sensitive so ON co.order_id = so.order_id
""").collect()

# Re-apply masking — CREATE OR REPLACE drops column policies
sf_session.sql(f"""
    ALTER ICEBERG TABLE {TBL_INSIGHTS}
    MODIFY COLUMN risk_level
    SET MASKING POLICY {SF_MANAGED_ICEBERG_DB}.{SF_DEMO_SCHEMA}.MASK_RISK_LEVEL
""").collect()

Step 5 — Governance Comparison via Snowpark Connect

COLS = ["order_id", "product", "amount", "region", "risk_level"]

switch_role("ACCOUNTADMIN")
admin_df = sf_session.sql(f"SELECT {','.join(COLS)} FROM {TBL_INSIGHTS}").to_pandas()
admin_df.columns = [c.lower() for c in admin_df.columns]

switch_role(SF_READER_ROLE)
reader_df = sf_session.sql(f"SELECT {','.join(COLS)} FROM {TBL_INSIGHTS}").to_pandas()
reader_df.columns = [c.lower() for c in reader_df.columns]

sf_session.sql().to_pandas() returns uppercase column names. Call .columns = [c.lower() for c in df.columns] immediately after to_pandas().

Scenario 2 — Cortex Analyst Demo

Demo Prompt Sequence

"What is the total revenue by risk level?"

"Which orders are classified as HIGH risk and what are their operational notes?"

"Show me all SHIPPED orders in the US-WEST region"

"Compare total revenue from enriched orders versus source federated orders"

"What is the average order value by fulfillment status?"

Conclusion And Resources

What You Built

What You Learned

Governance Summary

Key insight: Snowflake's Horizon governance and Databricks UC governance are completely independent. Snowflake enforces its own masking policies on federated tables at query time — the source catalog's policies do not propagate, and Snowflake's policies do not reach into the source catalog.

Related Resources