29 - BigQuery Optimization Techniques

# File: de-onboarding/partition_example.py
from google.cloud import bigquery  # Import BigQuery client
from typing import Dict, Any  # Type annotations

def create_partitioned_table(project_id: str, dataset_id: str, table_id: str) -> Dict[str, Any]:
    """Create a partitioned BigQuery table by date."""
    client = bigquery.Client(project=project_id)  # Initialize client
    table_ref = f"{project_id}.{dataset_id}.{table_id}"  # Full table path
    schema = [
        bigquery.SchemaField("product", "STRING"),
        bigquery.SchemaField("price", "FLOAT"),
        bigquery.SchemaField("quantity", "INTEGER"),
        bigquery.SchemaField("sale_date", "DATE")  # Date for partitioning
    ]

    table = bigquery.Table(table_ref, schema=schema)  # Define table
    table.time_partitioning = bigquery.TimePartitioning(
        type_=bigquery.TimePartitioningType.DAY,  # Daily partitions
        field="sale_date"  # Partition by sale_date
    )

    table = client.create_table(table, exists_ok=True)  # Create table
    print(f"Created partitioned table {table_ref}")  # Debug
    return {
        "table_id": table.table_id,
        "partitioning_type": table.time_partitioning.type_,
        "partitioning_field": table.time_partitioning.field
    }

# Example usage
if __name__ == "__main__":
    result = create_partitioned_table("my-project", "my_dataset", "sales_partitioned")
    print(result)

# Expected Output:
# Created partitioned table my-project.my_dataset.sales_partitioned
# {'table_id': 'sales_partitioned', 'partitioning_type': 'DAY', 'partitioning_field': 'sale_date'}

# File: de-onboarding/cluster_example.py
from google.cloud import bigquery  # Import BigQuery client
from typing import Dict, Any  # Type annotations

def create_clustered_table(project_id: str, dataset_id: str, table_id: str) -> Dict[str, Any]:
    """Create a partitioned and clustered BigQuery table."""
    client = bigquery.Client(project=project_id)  # Initialize client
    table_ref = f"{project_id}.{dataset_id}.{table_id}"  # Full table path
    schema = [
        bigquery.SchemaField("product", "STRING"),
        bigquery.SchemaField("price", "FLOAT"),
        bigquery.SchemaField("quantity", "INTEGER"),
        bigquery.SchemaField("sale_date", "DATE")
    ]

    table = bigquery.Table(table_ref, schema=schema)  # Define table
    table.time_partitioning = bigquery.TimePartitioning(
        type_=bigquery.TimePartitioningType.DAY,
        field="sale_date"
    )
    table.clustering_fields = ["product"]  # Cluster by product

    table = client.create_table(table, exists_ok=True)  # Create table
    print(f"Created clustered table {table_ref}")  # Debug
    return {
        "table_id": table.table_id,
        "partitioning_type": table.time_partitioning.type_,
        "clustering_fields": table.clustering_fields
    }

# Example usage
if __name__ == "__main__":
    result = create_clustered_table("my-project", "my_dataset", "sales_clustered")
    print(result)

# Expected Output:
# Created clustered table my-project.my_dataset.sales_clustered
# {'table_id': 'sales_clustered', 'partitioning_type': 'DAY', 'clustering_fields': ['product']}

# File: de-onboarding/materialized_view_example.py
from google.cloud import bigquery  # Import BigQuery client
from typing import Dict, Any  # Type annotations

def create_materialized_view(project_id: str, dataset_id: str, view_id: str, source_table: str) -> Dict[str, Any]:
    """Create a materialized view for sales aggregates."""
    client = bigquery.Client(project=project_id)  # Initialize client
    view_ref = f"{project_id}.{dataset_id}.{view_id}"  # View path
    query = f"""
    SELECT
        product,
        sale_date,
        SUM(price * quantity) AS total_sales,
        SUM(quantity) AS total_quantity
    FROM `{source_table}`
    GROUP BY product, sale_date
    """

    view = bigquery.Table(view_ref)  # Define view
    view.materialized_view = bigquery.MaterializedViewDefinition(
        query=query,
        enable_refresh=True,
        refresh_interval_ms=3600000  # Refresh hourly
    )

    view = client.create_table(view, exists_ok=True)  # Create view
    print(f"Created materialized view {view_ref}")  # Debug
    return {
        "view_id": view.table_id,
        "query": query.strip(),
        "refresh_interval_ms": view.materialized_view.refresh_interval_ms
    }

# Example usage
if __name__ == "__main__":
    result = create_materialized_view(
        "my-project", "my_dataset", "sales_mv",
        "my-project.my_dataset.sales_clustered"
    )
    print(result)

# Expected Output:
# Created materialized view my-project.my_dataset.sales_mv
# {'view_id': 'sales_mv', 'query': 'SELECT\n    product,\n    sale_date,\n    SUM(price * quantity) AS total_sales,\n    SUM(quantity) AS total_quantity\nFROM `my-project.my_dataset.sales_clustered`\nGROUP BY product, sale_date', 'refresh_interval_ms': 3600000}

# File: de-onboarding/query_tuning_example.py
from google.cloud import bigquery  # Import BigQuery client
from typing import Dict, Any  # Type annotations

def run_queries(project_id: str, dataset_id: str, table_id: str) -> Dict[str, Any]:
    """Compare inefficient and optimized queries."""
    client = bigquery.Client(project=project_id)  # Initialize client
    table_ref = f"{project_id}.{dataset_id}.{table_id}"  # Table path

    # Inefficient query: SELECT * with no partitioning
    inefficient_query = f"""
    SELECT *
    FROM `{table_ref}`
    WHERE sale_date = '2023-10-01'
    """
    job_config = bigquery.QueryJobConfig(dry_run=True)  # Dry run to estimate bytes
    job = client.query(inefficient_query, job_config=job_config)
    inefficient_bytes = job.total_bytes_processed

    # Optimized query: Select specific columns, use partition
    optimized_query = f"""
    SELECT product, price, quantity
    FROM `{table_ref}`
    WHERE sale_date = '2023-10-01'
    AND product = 'Halal Laptop'
    """
    job = client.query(optimized_query, job_config=job_config)
    optimized_bytes = job.total_bytes_processed

    print(f"Inefficient query scans {inefficient_bytes} bytes")  # Debug
    print(f"Optimized query scans {optimized_bytes} bytes")  # Debug
    return {
        "inefficient_bytes": inefficient_bytes,
        "optimized_bytes": optimized_bytes,
        "savings_ratio": inefficient_bytes / optimized_bytes if optimized_bytes > 0 else float("inf")
    }

# Example usage
if __name__ == "__main__":
    result = run_queries("my-project", "my_dataset", "sales_clustered")
    print(result)

# Expected Output (varies by data):
# Inefficient query scans 1048576 bytes
# Optimized query scans 131072 bytes
# {'inefficient_bytes': 1048576, 'optimized_bytes': 131072, 'savings_ratio': 8.0}

# File: de-onboarding/cost_example.py
from google.cloud import bigquery  # Import BigQuery client
from typing import Dict, Any  # Type annotations

def estimate_query_cost(project_id: str, dataset_id: str, table_id: str) -> Dict[str, Any]:
    """Estimate query cost using dry run."""
    client = bigquery.Client(project=project_id)  # Initialize client
    table_ref = f"{project_id}.{dataset_id}.{table_id}"  # Table path
    query = f"""
    SELECT product, SUM(price * quantity) AS total_sales
    FROM `{table_ref}`
    WHERE sale_date = '2023-10-01'
    GROUP BY product
    """

    job_config = bigquery.QueryJobConfig(dry_run=True)  # Dry run
    job = client.query(query, job_config=job_config)
    bytes_scanned = job.total_bytes_processed
    cost_usd = (bytes_scanned / 1e12) * 5.0  # $5 per TB

    print(f"Query scans {bytes_scanned} bytes, costs ${cost_usd:.4f}")  # Debug
    return {
        "bytes_scanned": bytes_scanned,
        "cost_usd": cost_usd
    }

# Example usage
if __name__ == "__main__":
    result = estimate_query_cost("my-project", "my_dataset", "sales_clustered")
    print(result)

# Expected Output (varies by data):
# Query scans 131072 bytes, costs $0.0006
# {'bytes_scanned': 131072, 'cost_usd': 0.00065536}

import pandas as pd
data = [
    {"product": "Halal Laptop", "price": 999.99, "quantity": 2},
    {"product": "Halal Mouse", "price": 24.99, "quantity": 10},
    {"product": "Halal Keyboard", "price": 49.99, "quantity": 5}
]
pd.DataFrame(data).to_csv("data/sales.csv", index=False)
print("Seeded data/sales.csv")

min_price: 10.0
max_quantity: 100
required_fields:
  - product
  - price
  - quantity
product_prefix: 'Halal'
max_decimals: 2

# File: de-onboarding/utils.py (from Chapter 3, extended)
from typing import Any, Dict  # Type annotations

def is_numeric(s: str, max_decimals: int = 2) -> bool:
    """Check if string is a decimal number with up to max_decimals."""
    parts = s.split(".")  # Split on decimal point
    if len(parts) != 2 or not parts[0].isdigit() or not parts[1].isdigit():
        return False  # Invalid format
    return len(parts[1]) <= max_decimals  # Check decimal places

def clean_string(s: Any) -> str:
    """Strip whitespace from string."""
    return str(s).strip()

def is_numeric_value(x: Any) -> bool:
    """Check if value is numeric."""
    return isinstance(x, (int, float))  # Return True for numeric types

def has_valid_decimals(x: Any, max_decimals: int) -> bool:
    """Check if value has valid decimal places."""
    return is_numeric(str(x), max_decimals)  # Use is_numeric

def apply_valid_decimals(x: Any, max_decimals: int) -> bool:
    """Apply has_valid_decimals to a value."""
    return has_valid_decimals(x, max_decimals)

def is_integer(x: Any) -> bool:
    """Check if value is an integer."""
    return str(x).isdigit()  # Return True for integer strings

# File: de-onboarding/optimize_sales.py
from google.cloud import bigquery  # Import BigQuery client
import pandas as pd  # For CSV processing
import yaml  # For YAML parsing
import json  # For JSON export
from typing import Dict, Any, Tuple  # Type annotations
from datetime import datetime  # For date handling
import os  # For file checks

def read_config(config_path: str) -> Dict[str, Any]:
    """Read YAML configuration."""
    print(f"Opening config: {config_path}")  # Debug
    with open(config_path, "r") as file:
        config = yaml.safe_load(file)
    print(f"Loaded config: {config}")  # Debug
    return config

def load_sales_to_bigquery(
    project_id: str, dataset_id: str, table_id: str, csv_path: str, config: Dict[str, Any]
) -> Tuple[bigquery.Table, int]:
    """Load sales CSV to BigQuery with validation, prioritizing partitioning."""
    client = bigquery.Client(project=project_id)
    table_ref = f"{project_id}.{dataset_id}.{table_id}"

    # Load and validate CSV
    df = pd.read_csv(csv_path)
    df = df.dropna(subset=["product"])
    df = df[df["product"].str.startswith(config["product_prefix"])]
    df = df[df["quantity"].apply(is_integer)]
    df["quantity"] = df["quantity"].astype(int)
    df = df[df["quantity"] <= config["max_quantity"]]
    df = df[df["price"].apply(is_numeric_value)]
    df = df[df["price"] > 0]
    df = df[df["price"] >= config["min_price"]]
    df = df[df["price"].apply(lambda x: apply_valid_decimals(x, config["max_decimals"]))]

    # Add sale_date
    df["sale_date"] = datetime(2023, 10, 1).date()  # Static date for demo

    # Define schema
    schema = [
        bigquery.SchemaField("product", "STRING"),
        bigquery.SchemaField("price", "FLOAT"),
        bigquery.SchemaField("quantity", "INTEGER"),
        bigquery.SchemaField("sale_date", "DATE")
    ]

    # Create partitioned table (primary optimization)
    table = bigquery.Table(table_ref, schema=schema)
    table.time_partitioning = bigquery.TimePartitioning(
        type_=bigquery.TimePartitioningType.DAY,
        field="sale_date"
    )
    # Optionally add clustering for enhanced performance
    table.clustering_fields = ["product"]
    table = client.create_table(table, exists_ok=True)

    # Load data
    job_config = bigquery.LoadJobConfig(
        schema=schema,
        write_disposition="WRITE_TRUNCATE"
    )
    job = client.load_table_from_dataframe(df, table_ref, job_config=job_config)
    job.result()  # Wait for completion
    print(f"Loaded {len(df)} rows to {table_ref}")  # Debug
    return table, len(df)

def create_sales_materialized_view(
    project_id: str, dataset_id: str, view_id: str, source_table: str
) -> bigquery.Table:
    """Create materialized view for sales aggregates (optional enhancement)."""
    client = bigquery.Client(project=project_id)
    view_ref = f"{project_id}.{dataset_id}.{view_id}"
    query = f"""
    SELECT
        product,
        sale_date,
        SUM(price * quantity) AS total_sales,
        SUM(quantity) AS total_quantity
    FROM `{source_table}`
    GROUP BY product, sale_date
    """

    view = bigquery.Table(view_ref)
    view.materialized_view = bigquery.MaterializedViewDefinition(
        query=query,
        enable_refresh=True,
        refresh_interval_ms=3600000
    )
    view = client.create_table(view, exists_ok=True)
    print(f"Created materialized view {view_ref}")  # Debug
    return view

def run_optimized_query(
    project_id: str, dataset_id: str, table_id: str
) -> Tuple[Dict[str, Any], int]:
    """Run optimized query and estimate cost, leveraging partitioning."""
    client = bigquery.Client(project=project_id)
    table_ref = f"{project_id}.{dataset_id}.{table_id}"
    query = f"""
    SELECT product, SUM(price * quantity) AS total_sales
    FROM `{table_ref}`
    WHERE sale_date = '2023-10-01'
    AND product = 'Halal Laptop'
    GROUP BY product
    """

    # Dry run for cost
    job_config = bigquery.QueryJobConfig(dry_run=True)
    dry_job = client.query(query, job_config=job_config)
    bytes_scanned = dry_job.total_bytes_processed
    cost_usd = (bytes_scanned / 1e12) * 5.0

    # Run query
    job = client.query(query)
    results = job.result()
    rows = [dict(row) for row in results]

    print(f"Query scans {bytes_scanned} bytes, costs ${cost_usd:.4f}")  # Debug
    return {
        "results": rows,
        "bytes_scanned": bytes_scanned,
        "cost_usd": cost_usd
    }, len(rows)

def export_results(results: Dict[str, Any], json_path: str) -> None:
    """Export results to JSON."""
    print(f"Writing to: {json_path}")  # Debug
    with open(json_path, "w") as file:
        json.dump(results, file, indent=2)
    print(f"Exported results to {json_path}")  # Debug

def main() -> None:
    """Main function to optimize sales data warehouse, prioritizing partitioning."""
    project_id = "my-project"  # Replace with your project ID
    dataset_id = "my_dataset"  # Replace with your dataset ID
    table_id = "sales_optimized"
    view_id = "sales_mv"
    csv_path = "data/sales.csv"
    config_path = "data/config.yaml"
    json_path = "data/optimization_results.json"

    # Load configuration
    config = read_config(config_path)

    # Load data with primary focus on partitioning
    table, row_count = load_sales_to_bigquery(project_id, dataset_id, table_id, csv_path, config)

    # Optionally enhance with materialized view
    view = create_sales_materialized_view(project_id, dataset_id, view_id, table.full_table_id)

    # Run optimized query leveraging partitioning
    query_results, result_count = run_optimized_query(project_id, dataset_id, table_id)

    # Prepare and export results
    results = {
        "table_id": table.table_id,
        "partitioning_field": table.time_partitioning.field,
        "clustering_fields": table.clustering_fields,
        "view_id": view.table_id,
        "row_count": row_count,
        "query_results": query_results["results"],
        "bytes_scanned": query_results["bytes_scanned"],
        "cost_usd": query_results["cost_usd"]
    }
    export_results(results, json_path)

    print("\nOptimization Report:")  # Summary
    print(f"Table: {table.full_table_id}")
    print(f"Rows Loaded: {row_count}")
    print(f"Materialized View: {view.full_table_id}")
    print(f"Query Results: {query_results['results']}")
    print(f"Bytes Scanned: {query_results['bytes_scanned']}")
    print(f"Cost: ${query_results['cost_usd']:.4f}")
    print("**Performance Benchmark**: Log execution times using `job.result().total_ms` to compare queries. For this dataset, an unoptimized `SELECT *` might take 500ms, while the optimized query leveraging partitioning takes ~50ms.")
    print("Partitioning reduced scanned data by ~90% for date-based queries, based on typical pruning for daily partitions.")
    print("**Query Plan Analysis**: Use `bq query --explain` to verify partitioning efficiency; look for ‘PARTITION_SCAN’ stages indicating pruning.")

if __name__ == "__main__":
    main()

{
  "table_id": "sales_optimized",
  "partitioning_field": "sale_date",
  "clustering_fields": ["product"],
  "view_id": "sales_mv",
  "row_count": 3,
  "query_results": [
    {
      "product": "Halal Laptop",
      "total_sales": 1999.98
    }
  ],
  "bytes_scanned": 131072,
  "cost_usd": 0.00065536
}

{'table_id': 'sales_part', 'partitioning_field': 'sale_date'}

{'table_id': 'sales_cluster', 'clustering_fields': ['product']}

{'view_id': 'sales_mv', 'refresh_interval_ms': 3600000}

{'bytes_scanned': 131072, 'cost_usd': 0.00065536}

from google.cloud import bigquery
def run_query(project_id: str, dataset_id: str, table_id: str) -> Dict[str, Any]:
    client = bigquery.Client(project=project_id)
    query = f"""
    SELECT *
    FROM `{project_id}.{dataset_id}.{table_id}`
    """
    job = client.query(query, job_config=bigquery.QueryJobConfig(dry_run=True))
    return {"bytes_scanned": job.total_bytes_processed}

{'bytes_scanned': 131072, 'cost_usd': 0.00065536}

from google.cloud import bigquery
from typing import Dict, Any

def create_partitioned_table(project_id: str, dataset_id: str, table_id: str) -> Dict[str, Any]:
    client = bigquery.Client(project=project_id)
    table_ref = f"{project_id}.{dataset_id}.{table_id}"
    schema = [
        bigquery.SchemaField("product", "STRING"),
        bigquery.SchemaField("price", "FLOAT"),
        bigquery.SchemaField("quantity", "INTEGER"),
        bigquery.SchemaField("sale_date", "DATE")
    ]
    table = bigquery.Table(table_ref, schema=schema)
    table.time_partitioning = bigquery.TimePartitioning(
        type_=bigquery.TimePartitioningType.DAY,
        field="product"  # Incorrect field
    )
    table = client.create_table(table, exists_ok=True)
    return {
        "table_id": table.table_id,
        "partitioning_field": table.time_partitioning.field
    }

{'table_id': 'sales_debug', 'partitioning_field': 'sale_date'}

from google.cloud import bigquery
from typing import Dict, Any

def create_partitioned_table(project_id: str, dataset_id: str, table_id: str) -> Dict[str, Any]:
    """Create a partitioned BigQuery table."""
    client = bigquery.Client(project=project_id)
    table_ref = f"{project_id}.{dataset_id}.{table_id}"
    schema = [
        bigquery.SchemaField("product", "STRING"),
        bigquery.SchemaField("price", "FLOAT"),
        bigquery.SchemaField("quantity", "INTEGER"),
        bigquery.SchemaField("sale_date", "DATE")
    ]
    table = bigquery.Table(table_ref, schema=schema)
    table.time_partitioning = bigquery.TimePartitioning(
        type_=bigquery.TimePartitioningType.DAY,
        field="sale_date"
    )
    table = client.create_table(table, exists_ok=True)
    return {
        "table_id": table.table_id,
        "partitioning_field": table.time_partitioning.field
    }

# Test
print(create_partitioned_table("my-project", "my_dataset", "sales_part"))

from google.cloud import bigquery
from typing import Dict, Any

def create_clustered_table(project_id: str, dataset_id: str, table_id: str) -> Dict[str, Any]:
    """Create a clustered BigQuery table."""
    client = bigquery.Client(project=project_id)
    table_ref = f"{project_id}.{dataset_id}.{table_id}"
    schema = [
        bigquery.SchemaField("product", "STRING"),
        bigquery.SchemaField("price", "FLOAT"),
        bigquery.SchemaField("quantity", "INTEGER"),
        bigquery.SchemaField("sale_date", "DATE")
    ]
    table = bigquery.Table(table_ref, schema=schema)
    table.clustering_fields = ["product"]
    table = client.create_table(table, exists_ok=True)
    return {
        "table_id": table.table_id,
        "clustering_fields": table.clustering_fields
    }

# Test
print(create_clustered_table("my-project", "my_dataset", "sales_cluster"))

from google.cloud import bigquery
from typing import Dict, Any

def create_materialized_view(project_id: str, dataset_id: str, view_id: str, source_table: str) -> Dict[str, Any]:
    """Create a materialized view."""
    client = bigquery.Client(project=project_id)
    view_ref = f"{project_id}.{dataset_id}.{view_id}"
    query = f"""
    SELECT
        product,
        sale_date,
        SUM(price * quantity) AS total_sales
    FROM `{source_table}`
    GROUP BY product, sale_date
    """
    view = bigquery.Table(view_ref)
    view.materialized_view = bigquery.MaterializedViewDefinition(
        query=query,
        enable_refresh=True,
        refresh_interval_ms=3600000
    )
    view = client.create_table(view, exists_ok=True)
    return {
        "view_id": view.table_id,
        "refresh_interval_ms": view.materialized_view.refresh_interval_ms
    }

# Test
print(create_materialized_view("my-project", "my_dataset", "sales_mv", "my-project.my_dataset.sales_optimized"))

from google.cloud import bigquery
from typing import Dict, Any

def run_optimized_query(project_id: str, dataset_id: str, table_id: str) -> Dict[str, Any]:
    """Run optimized query."""
    client = bigquery.Client(project=project_id)
    table_ref = f"{project_id}.{dataset_id}.{table_id}"
    query = f"""
    SELECT product, SUM(price * quantity) AS total_sales
    FROM `{table_ref}`
    WHERE sale_date = '2023-10-01'
    GROUP BY product
    """
    job = client.query(query, job_config=bigquery.QueryJobConfig(dry_run=True))
    bytes_scanned = job.total_bytes_processed
    cost_usd = (bytes_scanned / 1e12) * 5.0
    return {
        "bytes_scanned": bytes_scanned,
        "cost_usd": cost_usd
    }

# Test
print(run_optimized_query("my-project", "my_dataset", "sales_optimized"))

from google.cloud import bigquery
from typing import Dict, Any

def run_query(project_id: str, dataset_id: str, table_id: str) -> Dict[str, Any]:
    """Run optimized query."""
    client = bigquery.Client(project=project_id)
    query = f"""
    SELECT product, price, quantity
    FROM `{project_id}.{dataset_id}.{table_id}`
    WHERE sale_date = '2023-10-01'
    """
    job = client.query(query, job_config=bigquery.QueryJobConfig(dry_run=True))
    bytes_scanned = job.total_bytes_processed
    cost_usd = (bytes_scanned / 1e12) * 5.0
    return {"bytes_scanned": bytes_scanned, "cost_usd": cost_usd}

# Test
print(run_query("my-project", "my_dataset", "sales_optimized"))

def write_concepts():
    """Write partitioning vs. clustering explanation to file."""
    explanation = """
    Partitioning is best for date-based queries, like daily sales, as it splits data by sale_date, reducing scanned data (e.g., 1TB to 10GB for one day). Clustering is ideal for product-based queries, like Halal Laptop sales, as it organizes data within partitions by product, improving filter efficiency. Use both for datasets with mixed query patterns to optimize cost and performance.
    """
    follow_up = """
    Combining partitioning and clustering leverages both date-based and column-based pruning, optimizing mixed query patterns (e.g., daily Halal Laptop sales) for cost and speed in Hijra Group’s analytics.
    """
    with open("de-onboarding/ex6_concepts.txt", "w") as file:
        file.write(explanation.strip() + "\n" + follow_up.strip())
    print("Wrote explanation to de-onboarding/ex6_concepts.txt")

# Test
write_concepts()

from google.cloud import bigquery
import pandas as pd
from typing import None

def export_materialized_view(project_id: str, dataset_id: str, view_id: str, csv_path: str) -> None:
    """Export materialized view to CSV for visualization."""
    client = bigquery.Client(project=project_id)
    view_ref = f"{project_id}.{dataset_id}.{view_id}"
    query = f"SELECT * FROM `{view_ref}`"

    df = client.query(query).to_dataframe()
    df.to_csv(csv_path, index=False)
    assert len(pd.read_csv(csv_path)) == 3  # Validate row count
    print(f"CSV saved to {csv_path}")
    print(f"Row count: {len(pd.read_csv(csv_path))}")

# Test
export_materialized_view("my-project", "my_dataset", "sales_mv", "data/sales_mv.csv")

from google.cloud import bigquery
import pandas as pd
from typing import None

def log_query_costs(project_id: str, dataset_id: str, table_id: str, csv_path: str) -> None:
    """Log query costs to CSV for visualization."""
    client = bigquery.Client(project=project_id)
    table_ref = f"{project_id}.{dataset_id}.{table_id}"

    queries = [
        ("optimized", """
            SELECT product, SUM(price * quantity) AS total_sales
            FROM `{table_ref}`
            WHERE sale_date = '2023-10-01'
            GROUP BY product
        """),
        ("unoptimized", """
            SELECT *
            FROM `{table_ref}`
            WHERE sale_date = '2023-10-01'
        """)
    ]

    results = []
    for query_type, query in queries:
        job = client.query(query.format(table_ref=table_ref), job_config=bigquery.QueryJobConfig(dry_run=True))
        bytes_scanned = job.total_bytes_processed
        cost_usd = (bytes_scanned / 1e12) * 5.0
        results.append({"query_type": query_type, "bytes_scanned": bytes_scanned, "cost_usd": cost_usd})

    df = pd.DataFrame(results)
    df.to_csv(csv_path, index=False)
    assert len(pd.read_csv(csv_path)) >= 2  # Validate row count
    print(f"CSV saved to {csv_path}")
    print(f"Row count: {len(pd.read_csv(csv_path))}")

# Test
log_query_costs("my-project", "my_dataset", "sales_optimized", "data/query_costs.csv")

from google.cloud import bigquery
from typing import Dict, Any

def create_partitioned_table(project_id: str, dataset_id: str, table_id: str) -> Dict[str, Any]:
    """Create a partitioned BigQuery table."""
    client = bigquery.Client(project=project_id)
    table_ref = f"{project_id}.{dataset_id}.{table_id}"
    schema = [
        bigquery.SchemaField("product", "STRING"),
        bigquery.SchemaField("price", "FLOAT"),
        bigquery.SchemaField("quantity", "INTEGER"),
        bigquery.SchemaField("sale_date", "DATE")
    ]
    table = bigquery.Table(table_ref, schema=schema)
    table.time_partitioning = bigquery.TimePartitioning(
        type_=bigquery.TimePartitioningType.DAY,
        field="sale_date"  # Fixed from incorrect 'product'
    )
    table = client.create_table(table, exists_ok=True)
    return {
        "table_id": table.table_id,
        "partitioning_field": table.time_partitioning.field
    }

# Test
print(create_partitioned_table("my-project", "my_dataset", "sales_debug"))