SQL Transformations within the OVHcloud Data Platform#

# Import necessary modules from SDK
from forepaas.dwh import connect, bulk_insert
from forepaas.core.settings import CONFIG
from forepaas.dwh.logical import LogicalObject
import pandas as pd #useful for displaying dataframes

print("SDK modules imported successfully.")

# Connect to the Lakehouse Manager
# Connect to the default Lakehouse Manager dataset
connector = connect("dwh/default_dataset/")

print("Listing tables in 'dwh/default_dataset/':")
available_tables = connector.list()
for table_name in available_tables:
    print(f"- {table_name}")

if "dirty_cafe_sales" in available_tables:
    print("\n'dirty_cafe_sales' table found!")
else:
    print("\nWARNING: 'dirty_cafe_sales' table not found. Please check the table name or path.")

# Connect to the 'dirty_cafe_sales' table
df_raw_sales = connector.select("dirty_cafe_sales")

print("\nDataset information:")
df_raw_sales.info()

print("\nDataset details:")
df_raw_sales.describe()

print("\nSample of Raw Data (first 5 rows):")
display(df_raw_sales.head())

print("--- Running Simple SQL Query ---")

SIMPLE_SQL_QUERY = """
SELECT
    CAST(valid_transaction_date AS DATE) AS sale_date,
    -- Handle 'ERROR' and empty strings in location
    CASE
        WHEN location = 'ERROR' THEN 'Unknown'
        WHEN TRIM(location) = '' THEN 'Unknown'
        ELSE location
    END AS clean_location,
    SUM(TRY_CAST(total_spent AS DOUBLE)) AS gross_revenue_dirty
FROM
    (
        SELECT
            TRY_CAST(transaction_date AS DATE) AS valid_transaction_date,
            location,
            total_spent
        FROM
            dirty_cafe_sales
    ) AS subquery_sales
WHERE
    valid_transaction_date IS NOT NULL
GROUP BY
    CAST(valid_transaction_date AS DATE),
    CASE
        WHEN location = 'ERROR' THEN 'Unknown'
        WHEN TRIM(location) = '' THEN 'Unknown'
        ELSE location
    END
ORDER BY
    sale_date DESC, clean_location
""" # No semicolon at the end here!

try:
    # Execute the query using your connector's method.
    df_simple_result = connector.query(SIMPLE_SQL_QUERY)
    
    # --- Pandas Post-Processing for Data Types ---
    # Convert 'sale_date' to datetime objects for proper date operations
    df_simple_result['sale_date'] = pd.to_datetime(df_simple_result['sale_date'])
    # --- End Pandas Post-Processing ---

    print("\nSimple Query Results (first 10 rows):")
    display(df_simple_result.head(10))
    print("\nSimple Query Results Schema:")
    df_simple_result.info()

except Exception as e:
    print(f"Error executing Simple SQL query: {e}")
    print("\nFailed SQL Query:\n", SIMPLE_SQL_QUERY)

print("\n--- Running Complex SQL Transformation Query ---")

COMPLEX_SQL_TRANSFORMATION_QUERY = """
WITH cleaned_and_corrected_sales AS (
    SELECT
        transaction_id,
        item,
        -- Clean and cast quantity: 'UNKNOWN' becomes NULL, then cast to INTEGER.
        -- TRY_CAST handles non-numeric strings safely by returning NULL.
        TRY_CAST(NULLIF(quantity, 'UNKNOWN') AS INTEGER) AS quantity_cleaned,
        -- Ensure price_per_unit is numeric, handling potential non-numeric entries safely.
        TRY_CAST(price_per_unit AS DOUBLE) AS price_per_unit_cleaned,
        payment_method,
        -- Clean location: 'ERROR' and empty strings become 'Unknown'.
        CASE
            WHEN location = 'ERROR' THEN 'Unknown'
            WHEN TRIM(location) = '' THEN 'Unknown'
            ELSE location
        END AS clean_location,
        -- Use TRY_CAST for transaction_date to handle bad date strings safely, then filter later.
        TRY_CAST(transaction_date AS DATE) AS sale_date_raw
    FROM
        dirty_cafe_sales
),
final_calculated_sales AS (
    SELECT
        transaction_id,
        item,
        quantity_cleaned AS final_quantity,
        price_per_unit_cleaned AS final_price_per_unit,
        -- Recalculate total_spent based on cleaned quantity and price_per_unit.
        quantity_cleaned * price_per_unit_cleaned AS calculated_total_spent,
        payment_method,
        clean_location,
        sale_date_raw AS sale_date
    FROM
        cleaned_and_corrected_sales
    -- Filter out rows where crucial values (quantity, price_per_unit, or sale_date)
    -- couldn't be cleanly converted, ensuring only valid data proceeds.
    WHERE
        quantity_cleaned IS NOT NULL 
        AND price_per_unit_cleaned IS NOT NULL
        AND sale_date_raw IS NOT NULL 
)
SELECT
    sale_date,
    item,
    clean_location AS location,
    COUNT(DISTINCT transaction_id) AS number_of_transactions,
    SUM(final_quantity) AS total_items_sold,
    SUM(calculated_total_spent) AS total_revenue_cleaned,
    AVG(final_price_per_unit) AS average_item_price_per_unit,
    -- Pivot revenue by payment method using Trino's FILTER clause
    SUM(calculated_total_spent) FILTER (WHERE payment_method = 'Credit Card') AS revenue_credit_card,
    SUM(calculated_total_spent) FILTER (WHERE payment_method = 'Cash') AS revenue_cash,
    SUM(calculated_total_spent) FILTER (WHERE payment_method = 'Digital Wallet') AS revenue_digital_wallet,
    -- Calculate average quantity per transaction for this group
    CAST(SUM(final_quantity) AS DOUBLE) / CAST(COUNT(DISTINCT transaction_id) AS DOUBLE) AS avg_quantity_per_transaction
FROM
    final_calculated_sales
GROUP BY
    sale_date,
    item,
    clean_location
ORDER BY
    sale_date DESC, total_revenue_cleaned DESC
""" # No semicolon at the end here!

try:
    # Execute the complex transformation query
    df_transformed = connector.query(COMPLEX_SQL_TRANSFORMATION_QUERY)

    # --- Pandas Post-Processing for Data Types ---
    # Convert 'sale_date' to datetime objects for proper date operations
    df_transformed['sale_date'] = pd.to_datetime(df_transformed['sale_date'])
    # --- End Pandas Post-Processing ---

    print("\nTransformed Cafe Sales Data (Sample - first 5 rows):")
    display(df_transformed.head())
    print(f"\nTransformed Data Schema:")
    df_transformed.info()

    print(f"\nTotal rows in transformed data: {len(df_transformed)}")

except Exception as e:
    print(f"Error during Complex SQL transformation: {e}")
    print("\nFailed SQL Query:\n", COMPLEX_SQL_TRANSFORMATION_QUERY)

print("\n--- Creating Physical Table from Complex SQL Transformation Query ---")

# Define the name of your new cleaned table. This variable can be reused across cells.
NEW_CLEANED_TABLE_NAME = "cleaned_cafe_sales_daily_summary"

# 1. DROP TABLE statement (removes the table if it already exists, for idempotent runs)
# Note: No trailing semicolon at the very end of the string.
DROP_TABLE_QUERY = f"DROP TABLE IF EXISTS {NEW_CLEANED_TABLE_NAME}"

# 2. CREATE TABLE AS SELECT statement
# This uses the same logic from the COMPLEX_SQL_TRANSFORMATION_QUERY
# Note: No trailing semicolon at the very end of the string.
CTAS_CORE_QUERY = f"""
CREATE TABLE {NEW_CLEANED_TABLE_NAME} AS
WITH cleaned_and_corrected_sales AS (
    SELECT
        transaction_id,
        item,
        TRY_CAST(NULLIF(quantity, 'UNKNOWN') AS INTEGER) AS quantity_cleaned,
        TRY_CAST(price_per_unit AS DOUBLE) AS price_per_unit_cleaned,
        payment_method,
        CASE
            WHEN location = 'ERROR' THEN 'Unknown'
            WHEN TRIM(location) = '' THEN 'Unknown'
            ELSE location
        END AS clean_location,
        TRY_CAST(transaction_date AS DATE) AS sale_date_raw
    FROM
        dirty_cafe_sales
),
final_calculated_sales AS (
    SELECT
        transaction_id,
        item,
        quantity_cleaned AS final_quantity,
        price_per_unit_cleaned AS final_price_per_unit,
        quantity_cleaned * price_per_unit_cleaned AS calculated_total_spent,
        payment_method,
        clean_location,
        sale_date_raw AS sale_date
    FROM
        cleaned_and_corrected_sales
    WHERE
        quantity_cleaned IS NOT NULL 
        AND price_per_unit_cleaned IS NOT NULL
        AND sale_date_raw IS NOT NULL 
)
SELECT
    sale_date,
    item,
    clean_location AS location,
    COUNT(DISTINCT transaction_id) AS number_of_transactions,
    SUM(final_quantity) AS total_items_sold,
    SUM(calculated_total_spent) AS total_revenue_cleaned,
    AVG(final_price_per_unit) AS average_item_price_per_unit,
    SUM(calculated_total_spent) FILTER (WHERE payment_method = 'Credit Card') AS revenue_credit_card,
    SUM(calculated_total_spent) FILTER (WHERE payment_method = 'Cash') AS revenue_cash,
    SUM(calculated_total_spent) FILTER (WHERE payment_method = 'Digital Wallet') AS revenue_digital_wallet,
    CAST(SUM(final_quantity) AS DOUBLE) / CAST(COUNT(DISTINCT transaction_id) AS DOUBLE) AS avg_quantity_per_transaction
FROM
    final_calculated_sales
GROUP BY
    sale_date,
    item,
    clean_location
ORDER BY
    sale_date DESC, total_revenue_cleaned DESC
"""

try:
    # Execute DROP TABLE first
    print(f"Dropping table {NEW_CLEANED_TABLE_NAME} if it exists...")
    connector.query(DROP_TABLE_QUERY)
    print("Drop table command executed.")

    # Then execute CREATE TABLE AS SELECT
    print(f"Creating table {NEW_CLEANED_TABLE_NAME}...")
    # For DDL operations like CREATE TABLE, connector.query() might return an empty DataFrame or None.
    connector.query(CTAS_CORE_QUERY)
    
    print(f"\nSuccessfully created table: {NEW_CLEANED_TABLE_NAME}")
    
    # Optional: Verify the table was created by querying its schema or a few rows
    print(f"\nVerifying schema of new table: {NEW_CLEANED_TABLE_NAME}")
    df_verify = connector.query(f"SELECT * FROM {NEW_CLEANED_TABLE_NAME} LIMIT 5")
    display(df_verify)
    df_verify.info()

except Exception as e:
    print(f"Error during CTAS operation for {NEW_CLEANED_TABLE_NAME}: {e}")
    if "DROP TABLE" in str(e) and DROP_TABLE_QUERY in str(e):
        print("\nFailed SQL Query (DROP TABLE):\n", DROP_TABLE_QUERY)
    elif "CREATE TABLE" in str(e) and CTAS_CORE_QUERY in str(e):
        print("\nFailed SQL Query (CREATE TABLE AS SELECT):\n", CTAS_CORE_QUERY)
    else:
        print("\nFailed SQL Query:\n", e)

# You'll need to know your catalog and schema names.
# Example: your_catalog_name = "default_dataset", your_schema_name = "sales_data"
# To retrieve catalog and schema name you can execute the following commands in the Lakehouse Manager Explorer
# CATALOG LIST: show catalogs
# SCHEMA LIST: show schemas from {catalog_name}

your_catalog_name = "your_main_catalog" # <<< IMPORTANT: Replace with your actual catalog name!
your_schema_name = "your_schema_name"   # <<< IMPORTANT: Replace with your actual schema name!
your_table_name = "cleaned_cafe_sales_daily_summary"

print(f"\n--- Exploring the New Table: {your_table_name} ---")

try:
    # Describe your new table
    df_describe = connector.query(f"DESCRIBE {your_catalog_name}.{your_schema_name}.{your_table_name}")
    print(f"\nDescription of table '{your_table_name}':")
    display(df_describe)
    
    # Select some data from your new table
    df_sample_data = connector.query(f"SELECT * FROM {your_catalog_name}.{your_schema_name}.{your_table_name} LIMIT 5")
    print(f"\nSample data from '{your_table_name}':")
    display(df_sample_data)

except Exception as e:
    print(f"Error exploring table '{your_table_name}': {e}")
    print("Please check the full table path (catalog.schema.table) and permissions.")

print(f"\n--- Creating LogicalObject for '{NEW_CLEANED_TABLE_NAME}' ---")

try:
        
    logical_cleaned_sales = LogicalObject().create_from_physical(NEW_CLEANED_TABLE_NAME)
    print(f"Successfully created logical object for: {NEW_CLEANED_TABLE_NAME}")

except NameError:
    print("Error: 'LogicalObject' is not defined. Please ensure you have imported the correct library/class or that it's globally accessible.")
except Exception as e:
    print(f"Error creating logical object: {e}")