-- A retail team needs to filter sales by region across multiple reports
-- Instead of repeating the JOIN + WHERE in every query, wrap it in an iTVF

CREATE FUNCTION dbo.GetRegionalSales(@region NVARCHAR(50), @startDate DATE)
RETURNS TABLE
WITH SCHEMABINDING
AS
RETURN (
    SELECT 
        o.OrderID,
        o.CustomerID,
        o.TotalAmount,
        o.OrderDate,
        s.StoreName,
        s.Region
    FROM dbo.Orders AS o
    INNER JOIN dbo.Stores AS s ON o.StoreID = s.StoreID
    WHERE s.Region = @region
      AND o.OrderDate >= @startDate
);

-- Usage: analysts can now query regionally without knowing the join logic
SELECT * FROM dbo.GetRegionalSales('West', '2025-01-01')
WHERE TotalAmount > 500
ORDER BY OrderDate DESC;

-- E-commerce team applies tiered discounts based on customer loyalty level
-- This logic is reused across order processing, reporting, and forecasting

CREATE FUNCTION dbo.CalculateDiscount(
    @basePrice DECIMAL(12,2),
    @loyaltyTier INT
)
RETURNS DECIMAL(12,2)
AS
BEGIN
    RETURN @basePrice * (
        CASE 
            WHEN @loyaltyTier >= 4 THEN 0.80  -- 20% off for platinum
            WHEN @loyaltyTier = 3  THEN 0.85  -- 15% off for gold
            WHEN @loyaltyTier = 2  THEN 0.90  -- 10% off for silver
            ELSE 1.00                          -- no discount
        END
    );
END;

-- Usage in a report query
SELECT 
    ProductName,
    ListPrice,
    dbo.CalculateDiscount(ListPrice, c.LoyaltyTier) AS FinalPrice
FROM dbo.Products p
INNER JOIN dbo.Customers c ON p.LastBuyerID = c.CustomerID;

SELECT 
    b.name AS function_name, 
    a.is_inlineable
FROM sys.sql_modules a
INNER JOIN sys.objects b ON a.object_id = b.object_id
WHERE b.type = 'FN';

-- If is_inlineable = 0, the UDF will FAIL when called on user tables

-- SaaS application calculates customer health score from multiple signals
-- Multi-statement TVF aggregates across tables — only works in Fabric SQL Database

CREATE FUNCTION dbo.GetCustomerHealthScores(@tenantId INT)
RETURNS @results TABLE (
    CustomerID INT,
    HealthScore DECIMAL(5,2),
    RiskLevel NVARCHAR(20),
    LastActivity DATETIME2
)
AS
BEGIN
    INSERT INTO @results
    SELECT 
        c.CustomerID,
        -- Weighted score from engagement, support, and usage signals
        (c.EngagementScore * 0.4) + (c.NPS * 0.3) + (c.UsageIndex * 0.3) AS HealthScore,
        CASE 
            WHEN (c.EngagementScore * 0.4) + (c.NPS * 0.3) + (c.UsageIndex * 0.3) > 80 THEN 'Healthy'
            WHEN (c.EngagementScore * 0.4) + (c.NPS * 0.3) + (c.UsageIndex * 0.3) > 50 THEN 'At Risk'
            ELSE 'Critical'
        END,
        c.LastActivityDate
    FROM dbo.CustomerMetrics c
    WHERE c.TenantID = @tenantId
      AND c.IsActive = 1;
    RETURN;
END;

-- Used by the app's customer success dashboard
SELECT * FROM dbo.GetCustomerHealthScores(42) WHERE RiskLevel = 'Critical';

import fabric.functions as fn

udf = fn.UserDataFunctions()

@udf.function()
def standardize_category(productName: str, rawCategory: str) -> dict:
    """
    Retail team receives product data from multiple suppliers with inconsistent
    categories. This function normalizes them into the company's standard taxonomy.
    Called from data pipelines during ingestion and from Power Apps for manual review.
    """
    category_mapping = {
        "electronics": ["tech", "devices", "gadgets", "computing", "hardware"],
        "clothing": ["apparel", "fashion", "garments", "wear", "textiles"],
        "home": ["furniture", "household", "kitchen", "decor", "garden"],
        "food": ["grocery", "beverages", "snacks", "organic", "fresh"]
    }
    
    raw_lower = rawCategory.lower().strip()
    standardized = next(
        (standard for standard, aliases in category_mapping.items() 
         if raw_lower in aliases),
        "other"
    )
    
    return {
        "product_name": productName,
        "original_category": rawCategory,
        "standardized_category": standardized,
        "confidence": "high" if standardized != "other" else "needs_review"
    }

import fabric.functions as fn
from datetime import datetime

udf = fn.UserDataFunctions()

@udf.function()
def detect_sensor_anomaly(
    sensorId: str, 
    temperature: float, 
    pressure: float, 
    humidity: float
) -> dict:
    """
    Manufacturing plant monitors equipment via IoT sensors. This function
    is triggered by Fabric Activator when new readings arrive, flagging
    anomalies that need operator attention before equipment damage occurs.
    """
    alerts = []
    severity = "normal"
    
    # Temperature thresholds for industrial equipment
    if temperature > 85.0:
        alerts.append(f"CRITICAL: Temperature {temperature}°C exceeds safe limit (85°C)")
        severity = "critical"
    elif temperature > 70.0:
        alerts.append(f"WARNING: Temperature {temperature}°C approaching limit")
        severity = "warning"
    
    # Pressure differential check
    if pressure < 20.0 or pressure > 150.0:
        alerts.append(f"ALERT: Pressure {pressure} PSI outside normal range (20-150)")
        severity = "critical" if severity != "critical" else severity
    
    # Humidity correlation (high humidity + high temp = corrosion risk)
    if humidity > 80.0 and temperature > 60.0:
        alerts.append("WARNING: High humidity + temperature — corrosion risk")
        severity = max(severity, "warning", key=lambda x: ["normal","warning","critical"].index(x))
    
    return {
        "sensor_id": sensorId,
        "timestamp": datetime.utcnow().isoformat(),
        "severity": severity,
        "alerts": alerts,
        "requires_action": severity in ("warning", "critical")
    }

# Call the function from any Fabric notebook — no Spark cluster needed for the UDF itself
myFunctions = notebookutils.udf.getFunctions("ProductCategorizer")

# Single invocation
result = myFunctions.standardize_category(
    productName="Wireless Earbuds Pro",
    rawCategory="gadgets"
)
print(result)
# {'product_name': 'Wireless Earbuds Pro', 'standardized_category': 'electronics', 'confidence': 'high'}

# Cross-workspace access (e.g., shared library workspace)
shared = notebookutils.udf.getFunctions("ProductCategorizer", "shared-workspace-id")
result = shared.standardize_category(productName="Desk Lamp", rawCategory="decor")

Pipeline Design:
┌─────────────────┐     ┌──────────────────────┐     ┌──────────────────┐
│  Copy Activity  │────▶│  Functions Activity   │────▶│  Write to Lake   │
│  (Raw supplier  │     │  standardize_category │     │  (Curated layer) │
│   catalog CSV)  │     │  for each row         │     │                  │
└─────────────────┘     └──────────────────────┘     └──────────────────┘

The Functions Activity calls your User Data Function as a pipeline step,
allowing you to embed custom logic without spinning up a Spark job.

import fabric.functions as fn
import requests

udf = fn.UserDataFunctions()

@udf.function()
@udf.connection(alias="weather-api", argName="apiConnection")
@udf.generic_connection(argName="keyVault", audienceType="KeyVault")
def enrich_with_weather(
    city: str, 
    date: str, 
    apiConnection: fn.ConnectionInfo,
    keyVault: fn.ConnectionInfo
) -> dict:
    """
    Supply chain team enriches delivery forecasts with weather data.
    API key stored in Azure Key Vault, accessed via managed connection.
    """
    api_key = apiConnection.get_secret("api-key")
    response = requests.get(
        f"https://api.weather.example.com/v1/forecast",
        params={"city": city, "date": date},
        headers={"Authorization": f"Bearer {api_key}"}
    )
    data = response.json()
    
    return {
        "city": city,
        "date": date,
        "condition": data.get("condition", "unknown"),
        "risk_level": "high" if data.get("severe_weather") else "normal"
    }

from pyspark.sql.functions import udf
from pyspark.sql.types import StringType
import hashlib

@udf(returnType=StringType())
def mask_email(email):
    """
    Healthcare company masks patient emails in the bronze-to-silver layer.
    The masked value is deterministic (same input = same hash) so downstream 
    joins still work, but the original email cannot be recovered.
    """
    if email is None:
        return None
    local, domain = email.split("@")
    hashed = hashlib.sha256(local.encode()).hexdigest()[:8]
    return f"{hashed}@{domain}"

# Apply during bronze → silver transformation
df_patients = spark.read.format("delta").load("Tables/bronze_patients")
df_masked = df_patients.withColumn("email", mask_email("email"))
df_masked.write.format("delta").mode("overwrite").save("Tables/silver_patients")

import pandas as pd
from pyspark.sql.functions import pandas_udf
from pyspark.sql.types import DoubleType

@pandas_udf(DoubleType())
def rolling_avg_spend(amounts: pd.Series) -> pd.Series:
    """
    Financial services company calculates rolling average transaction amounts
    for fraud detection features. Pandas UDF makes this 50x faster than 
    row-by-row processing on 100M+ transaction records.
    """
    return amounts.rolling(window=7, min_periods=1).mean()

# Apply to transaction history for ML feature store
df_transactions = spark.read.format("delta").load("Tables/transactions")
df_features = df_transactions.withColumn(
    "rolling_avg_7d", 
    rolling_avg_spend("amount")
)
df_features.write.format("delta").mode("overwrite").save("Tables/ml_features")

import pandas as pd
from pyspark.sql.functions import pandas_udf
from pyspark.sql.types import StringType
import re

@pandas_udf(StringType())
def normalize_address(addresses: pd.Series) -> pd.Series:
    """
    Logistics company normalizes millions of shipping addresses to reduce 
    duplicate deliveries. Vectorized processing handles 50M rows in minutes.
    """
    def clean(addr):
        if addr is None:
            return None
        addr = addr.upper().strip()
        addr = re.sub(r'\bSTREET\b', 'ST', addr)
        addr = re.sub(r'\bAVENUE\b', 'AVE', addr)
        addr = re.sub(r'\bBOULEVARD\b', 'BLVD', addr)
        addr = re.sub(r'\bAPARTMENT\b', 'APT', addr)
        addr = re.sub(r'\s+', ' ', addr)
        return addr
    
    return addresses.apply(clean)

df_shipments = spark.read.format("delta").load("Tables/raw_shipments")
df_clean = df_shipments.withColumn("normalized_address", normalize_address("shipping_address"))

# Project structure for a shared UDF library
my_fabric_udfs/
├── setup.py
├── my_udfs/
│   ├── __init__.py
│   ├── masking.py        # mask_email, mask_phone, mask_ssn
│   ├── validation.py     # validate_email, validate_phone
│   └── enrichment.py     # geocode_address, currency_convert
└── tests/
    ├── test_masking.py
    └── test_validation.py

# Build the wheel
python setup.py bdist_wheel

# Upload the .whl to your Fabric Environment → Custom Libraries
# All notebooks using that environment can then:
from my_udfs.masking import mask_email
from my_udfs.validation import validate_email

START: What's your primary need?
│
├─ "Transform data inside a SQL query" 
│   ├─ Returns a table? ──────────────▶ T-SQL Inline TVF (GA)
│   └─ Returns a single value? ───────▶ T-SQL Scalar UDF (Preview, inlineable only)
│
├─ "Share reusable logic across multiple Fabric items"
│   ├─ Need REST endpoint? ───────────▶ User Data Function (Preview)
│   ├─ Need Activator trigger? ───────▶ User Data Function (Preview)
│   └─ Only notebooks? ──────────────▶ Environment + .whl package (GA)
│
├─ "Process millions of rows in a Spark pipeline"
│   ├─ Simple row logic? ─────────────▶ Standard PySpark UDF (GA)
│   └─ Vectorizable computation? ─────▶ Pandas UDF (GA, 10-100x faster)
│
└─ "Full T-SQL function support (MSTVF, triggers, etc.)"
    └─ ───────────────────────────────▶ Fabric SQL Database (GA, full Azure SQL parity)