-- Step 1: Create filter function
CREATE FUNCTION Security.fn_filter(@SalesRep varchar(50))
RETURNS TABLE WITH SCHEMABINDING AS
RETURN SELECT 1 AS result
WHERE @SalesRep = USER_NAME() OR USER_NAME() = 'manager@contoso.com';

-- Step 2: Create security policy
CREATE SECURITY POLICY SalesFilter
ADD FILTER PREDICATE Security.fn_filter(SalesPersonColumn)
ON dbo.Sales WITH (STATE = ON);

from pyspark.sql import functions as F

# Create hash of all tracked columns
df_source = df_source.withColumn(
    "row_hash",
    F.md5(F.concat_ws("||", F.col("col1"), F.col("col2"), F.col("col3")))
)
# Compare hash with target to detect changes
changed = df_source.join(df_target, "id").filter(
    df_source.row_hash != df_target.row_hash
)

MERGE INTO target USING source ON target.id = source.id
WHEN MATCHED THEN
    UPDATE SET target.name = source.name
WHEN NOT MATCHED THEN
    INSERT (id, name) VALUES (source.id, source.name)
WHEN NOT MATCHED BY SOURCE THEN DELETE;
-- Note: Fabric enables Low Shuffle Merge by default — no code changes needed

OPTIMIZE table_name;                           -- bin-packing (compacts small files)
OPTIMIZE table_name ZORDER BY (customer_id);   -- co-locate related data for filter queries
OPTIMIZE table_name VORDER;                    -- apply V-Order encoding

VACUUM table_name RETAIN 168 HOURS;  -- default 7 days = 168 hours
-- Removes data files no longer referenced older than retention period
-- Does NOT remove _delta_log
-- After VACUUM: time travel to vacuumed versions is IMPOSSIBLE

SELECT * FROM my_table VERSION AS OF 5;
SELECT * FROM my_table TIMESTAMP AS OF '2024-01-15T10:30:00Z';
DESCRIBE HISTORY my_table;
RESTORE TABLE my_table TO VERSION AS OF 3;

-- Warehouse syntax:
SELECT * FROM dbo.dim_customer
OPTION (FOR TIMESTAMP AS OF '2024-05-23T08:00:00');

# Schema evolution — opt-in per write operation
df.write.option("mergeSchema", "true").mode("append").saveAsTable("target")

# Complete schema replacement (use rarely)
df.write.option("overwriteSchema", "true").mode("overwrite").saveAsTable("target")

# For MERGE with schema evolution:
-- MERGE WITH SCHEMA EVOLUTION INTO target USING source ON ...

// where — filter rows (put FIRST for best performance)
StormEvents | where State == "FLORIDA" | where DamageProperty > 1000000

// project — select columns (REDUCES output; more efficient than extend)
StormEvents | project State, EventType, DamageProperty

// extend — add computed columns (KEEPS all existing columns)
StormEvents | extend TotalInjuries = InjuriesDirect + InjuriesIndirect

// summarize — aggregation (replaces GROUP BY)
StormEvents | summarize EventCount = count(), AvgDamage = avg(DamageProperty) by State
StormEvents | summarize count() by bin(StartTime, 1h)   // time-bucket (1 hour bins)

// join — combine datasets (filter BEFORE join for performance!)
Table1 | where Region == "US"
       | join kind=inner (Table2) on CommonColumn

// sort by — ordering (default is DESCENDING)
StormEvents | sort by DamageProperty desc | take 10

// render — MUST BE LAST
StormEvents | summarize count() by State
            | render barchart with (title="Events by State")

from pyspark.sql import functions as F
from pyspark.sql.window import Window

# Read Delta table
df = spark.read.format("delta").load("Tables/sales")
df = spark.table("lakehouse.sales")

# Common transforms
df = df.filter(F.col("amount") > 100)
df = df.withColumn("year", F.year(F.col("order_date")))
df = df.groupBy("category").agg(F.sum("amount").alias("total"))
df = df.dropDuplicates(["id", "date"])
df = df.fillna({"amount": 0, "category": "Unknown"})

# Window function (rank)
w = Window.partitionBy("region").orderBy(F.desc("sales"))
df = df.withColumn("rank", F.dense_rank().over(w))

# Write managed Delta table (visible in SQL endpoint)
df.write.mode("overwrite").saveAsTable("gold_sales")
df.write.mode("append").saveAsTable("gold_sales")

# Write with schema evolution
df.write.option("mergeSchema", "true").mode("append").saveAsTable("gold_sales")

-- COPY INTO (bulk load)
COPY INTO dbo.fact_sales
FROM 'https://storage.dfs.core.windows.net/container/data/*.parquet'
WITH (FILE_TYPE = 'PARQUET');

-- Zero-copy clone (snapshot for testing)
CREATE TABLE dbo.fact_sales_backup AS CLONE OF dbo.fact_sales;

-- Column-level security
GRANT SELECT ON dbo.customers (CustomerName, Region) TO [analyst@contoso.com];

-- Dynamic data masking
ALTER TABLE dbo.customers
ALTER COLUMN Email ADD MASKED WITH (FUNCTION = 'email()');

-- Cross-database query (same workspace only)
INSERT INTO WarehouseA.dbo.Table
SELECT * FROM LakehouseB.dbo.Table;

-- Time travel
SELECT * FROM dbo.dim_customer
OPTION (FOR TIMESTAMP AS OF '2024-05-23T08:00:00');