-- Create a distributed table from staging
CREATE TABLE [dbo].[FactSales]
WITH
(
    DISTRIBUTION = HASH(SaleDate),
    CLUSTERED COLUMNSTORE INDEX,
    PARTITION = (SaleDate RANGE RIGHT FOR VALUES
        ('2024-01-01', '2024-02-01', '2024-03-01',
         '2024-04-01', '2024-05-01', '2024-06-01'))
)
AS
SELECT
    s.SaleID,
    s.CustomerKey,
    p.ProductKey,
    s.SaleDate,
    s.Quantity,
    s.UnitPrice,
    s.Quantity * s.UnitPrice AS TotalAmount,
    d.DateKey,
    c.CustomerSegment
FROM [staging].[Sales] s
INNER JOIN [dim].[Customers] c ON s.CustomerID = c.CustomerID
INNER JOIN [dim].[Products] p ON s.ProductID = p.ProductID
INNER JOIN [dim].[Dates] d ON s.SaleDate = d.FullDate
WHERE s.SaleDate >= '2024-01-01';

-- Verify distribution
DBCC PDW_SHOWSPACEUSED('dbo.FactSales');

-- Create statistics for better query plans
CREATE STATISTICS STAT_FactSales_SaleDate
ON [dbo].[FactSales](SaleDate);

CREATE STATISTICS STAT_FactSales_CustomerKey
ON [dbo].[FactSales](CustomerKey);

-- Create indexed view for common aggregations
CREATE VIEW [dbo].[vw_DailySalesSummary]
WITH SCHEMABINDING
AS
SELECT
    SaleDate,
    COUNT_BIG(*) AS TotalTransactions,
    SUM(TotalAmount) AS DailyRevenue
FROM [dbo].[FactSales]
GROUP BY SaleDate;

CREATE UNIQUE CLUSTERED INDEX IX_vw_DailySalesSummary
ON [dbo].[vw_DailySalesSummary](SaleDate);

-- Create external data source pointing to ADLS
CREATE EXTERNAL DATA SOURCE [AzureDataLake]
WITH (
    LOCATION = 'https://stdatalake001.dfs.core.windows.net',
    CREDENTIAL = [ManagedIdentityCredential]
);

-- Create external file format for Parquet
CREATE EXTERNAL FILE FORMAT [ParquetFormat]
WITH (
    FORMAT_TYPE = PARQUET,
    DATA_COMPRESSION = 'org.apache.hadoop.io.compress.SnappyCodec'
);

-- Create external table
CREATE EXTERNAL TABLE [dbo].[ExternalSales]
WITH (
    LOCATION = 'curated/sales/',
    DATA_SOURCE = [AzureDataLake],
    FILE_FORMAT = [ParquetFormat]
)
AS
SELECT * FROM OPENROWSET(
    BULK 'curated/sales/**/*.parquet',
    FORMAT = 'PARQUET'
) WITH (
    SaleID BIGINT,
    CustomerKey INT,
    ProductKey INT,
    SaleDate DATE,
    Quantity INT,
    UnitPrice DECIMAL(18,2),
    TotalAmount DECIMAL(18,2)
) AS [Sales];

-- Query with pushdown computation
SELECT
    SaleDate,
    SUM(TotalAmount) AS Revenue,
    COUNT(*) AS Transactions
FROM [dbo].[ExternalSales]
WHERE SaleDate >= '2024-01-01'
GROUP BY SaleDate
ORDER BY SaleDate;

from azure.identity import DefaultAzureCredential
from azure.synapse.artifacts import ArtifactsClient
from azure.synapse.spark import SparkClient
import time

credential = DefaultAzureCredential()

# Artifacts Client
artifacts_client = ArtifactsClient(
    credential=credential,
    endpoint="https://syn-workspace.dev.azuresynapse.net"
)

# Run a SQL script
script_run = artifacts_client.sql_script.create_sql_script(
    sql_script_name="daily_etl",
    properties={
        "content": {
            "query": "EXEC sp_DailyETL @date = '2024-01-15'",
            "currentConnection": {
                "name": "Built-in"
            }
        }
    }
)

# Submit Spark job
spark_client = SparkClient(
    credential=credential,
    endpoint="https://syn-workspace.dev.azuresynapse.net",
    spark_pool_name="SparkPool01"
)

spark_client.spark_batch.create_spark_batch_job(
    spark_batch_job={
        "file": "abfss://notebooks@stdatalake001.dfs.core.windows.net/etl_job.py",
        "configuration": {
            "spark.dynamicAllocation.enabled": "true",
            "spark.dynamicAllocation.minExecutors": "1",
            "spark.dynamicAllocation.maxExecutors": "10"
        }
    }
)