48 - PostgreSQL Performance Optimization

from typing import Any  # Import Any for flexible type annotations
import psycopg2  # Import psycopg2 for PostgreSQL connectivity
from psycopg2.extras import DictCursor  # Import DictCursor for dictionary-like row access

# Connect to PostgreSQL database with connection parameters
conn: psycopg2.connection = psycopg2.connect(
    dbname="postgres",  # Database name
    user="postgres",  # Username
    password="postgres",  # Password
    host="localhost",  # Host address
    port="5432"  # Port number
)
cursor: DictCursor = conn.cursor(cursor_factory=DictCursor)  # Create cursor for dictionary-based results

# Create tasks table if it doesn't exist
cursor.execute("""
CREATE TABLE IF NOT EXISTS tasks (
    task_id TEXT PRIMARY KEY,  -- Unique identifier for each task
    description TEXT,  -- Task description
    status TEXT  -- Task status (e.g., Completed, Pending)
)
""")

# Create B-tree index on status column to speed up status-based queries
cursor.execute("CREATE INDEX IF NOT EXISTS idx_status ON tasks (status)")
conn.commit()  # Commit changes to the database

# Insert sample task data for testing
cursor.executemany(
    "INSERT INTO tasks (task_id, description, status) VALUES (%s, %s, %s)",
    [
        ("T001", "Process Halal Laptop sales", "Completed"),  # Task 1
        ("T002", "Validate Halal Mouse inventory", "Pending"),  # Task 2
        ("T003", "Update Halal Keyboard pricing", "In Progress")  # Task 3
    ]
)
conn.commit()  # Commit inserted data

# Query tasks by status to test index performance
cursor.execute("SELECT * FROM tasks WHERE status = %s", ("Completed",))
results: list[dict[str, Any]] = [dict(row) for row in cursor.fetchall()]  # Convert rows to dictionaries
print("Tasks with status 'Completed':", results)  # Display results

# Close cursor and connection to free resources
cursor.close()
conn.close()

# Expected Output:
# Tasks with status 'Completed': [{'task_id': 'T001', 'description': 'Process Halal Laptop sales', 'status': 'Completed'}]

from typing import Any  # Import Any for flexible type annotations
import psycopg2  # Import psycopg2 for PostgreSQL connectivity
from psycopg2.extras import DictCursor  # Import DictCursor for dictionary-like row access

# Connect to PostgreSQL database
conn: psycopg2.connection = psycopg2.connect(
    dbname="postgres",  # Database name
    user="postgres",  # Username
    password="postgres",  # Password
    host="localhost",  # Host address
    port="5432"  # Port number
)
cursor: DictCursor = conn.cursor(cursor_factory=DictCursor)  # Create cursor for dictionary-based results

# Create composite index on status and task_id for multi-condition queries
cursor.execute("CREATE INDEX IF NOT EXISTS idx_status_task_id ON tasks (status, task_id)")
conn.commit()  # Commit index creation

# Query tasks with multiple conditions to leverage composite index
cursor.execute("SELECT * FROM tasks WHERE status = %s AND task_id LIKE %s", ("Pending", "T%"))
results: list[dict[str, Any]] = [dict(row) for row in cursor.fetchall()]  # Convert rows to dictionaries
print("Filtered Tasks:", results)  # Display filtered tasks

# Close cursor and connection to free resources
cursor.close()
conn.close()

# Expected Output:
# Filtered Tasks: [{'task_id': 'T002', 'description': 'Validate Halal Mouse inventory', 'status': 'Pending'}]

from typing import Any  # Import Any for flexible type annotations
import psycopg2  # Import psycopg2 for PostgreSQL connectivity
from psycopg2.extras import DictCursor  # Import DictCursor for dictionary-like row access

# Connect to PostgreSQL database
conn: psycopg2.connection = psycopg2.connect(
    dbname="postgres",  # Database name
    user="postgres",  # Username
    password="postgres",  # Password
    host="localhost",  # Host address
    port="5432"  # Port number
)
cursor: DictCursor = conn.cursor(cursor_factory=DictCursor)  # Create cursor for dictionary-based results

# Analyze query plan for status-based query
cursor.execute("EXPLAIN SELECT * FROM tasks WHERE status = %s", ("Completed",))
plan: list[dict[str, Any]] = [dict(row) for row in cursor.fetchall()]  # Convert plan rows to dictionaries
print("Query Plan:", plan)  # Display query plan

# Close cursor and connection to free resources
cursor.close()
conn.close()

# Expected Output (simplified):
# Query Plan: [{'QUERY PLAN': 'Index Scan using idx_status on tasks ...'}]

from typing import Any  # Import Any for flexible type annotations
import psycopg2  # Import psycopg2 for PostgreSQL connectivity
from psycopg2.extras import DictCursor  # Import DictCursor for dictionary-like row access

# Connect to PostgreSQL database
conn: psycopg2.connection = psycopg2.connect(
    dbname="postgres",  # Database name
    user="postgres",  # Username
    password="postgres",  # Password
    host="localhost",  # Host address
    port="5432"  # Port number
)
cursor: DictCursor = conn.cursor(cursor_factory=DictCursor)  # Create cursor for dictionary-based results

# Analyze inefficient query plan (uses Seq Scan due to UPPER function)
cursor.execute("EXPLAIN SELECT * FROM tasks WHERE UPPER(status) = %s", ("COMPLETED",))
inefficient_plan: list[dict[str, Any]] = [dict(row) for row in cursor.fetchall()]  # Convert plan to dictionaries
print("Inefficient Plan:", inefficient_plan)  # Display inefficient plan

# Analyze optimized query plan (uses Index Scan)
cursor.execute("EXPLAIN SELECT * FROM tasks WHERE status = %s", ("Completed",))
optimized_plan: list[dict[str, Any]] = [dict(row) for row in cursor.fetchall()]  # Convert plan to dictionaries
print("Optimized Plan:", optimized_plan)  # Display optimized plan

# Close cursor and connection to free resources
cursor.close()
conn.close()

# Expected Output (simplified):
# Inefficient Plan: [{'QUERY PLAN': 'Seq Scan on tasks ...'}]
# Optimized Plan: [{'QUERY PLAN': 'Index Scan using idx_status on tasks ...'}]

from typing import Any  # Import Any for flexible type annotations
import psycopg2  # Import psycopg2 for PostgreSQL connectivity
from psycopg2.extras import DictCursor  # Import DictCursor for dictionary-like row access
import time  # Import time for measuring execution duration

# Connect to PostgreSQL database
conn: psycopg2.connection = psycopg2.connect(
    dbname="postgres",  # Database name
    user="postgres",  # Username
    password="postgres",  # Password
    host="localhost",  # Host address
    port="5432"  # Port number
)
cursor: DictCursor = conn.cursor(cursor_factory=DictCursor)  # Create cursor for dictionary-based results

# Record start time for query execution
start_time: float = time.time()
# Execute query to fetch tasks with specific status
cursor.execute("SELECT * FROM tasks WHERE status = %s", ("Completed",))
results: list[dict[str, Any]] = [dict(row) for row in cursor.fetchall()]  # Convert rows to dictionaries
# Record end time and calculate duration
end_time: float = time.time()
execution_time: float = end_time - start_time  # Calculate query execution time
print(f"Query Time: {execution_time:.4f} seconds")  # Display execution time
print("Results:", results)  # Display query results

# Close cursor and connection to free resources
cursor.close()
conn.close()

# Expected Output (example):
# Query Time: 0.0012 seconds
# Results: [{'task_id': 'T001', 'description': 'Process Halal Laptop sales', 'status': 'Completed'}]

CREATE TABLE tasks (
    task_id TEXT,
    description TEXT,
    status TEXT
);
INSERT INTO tasks (task_id, description, status) VALUES
('T001', 'Process Halal Laptop sales', 'Completed'),
('T002', 'Validate Halal Mouse inventory', 'Pending'),
('T003', 'Update Halal Keyboard pricing', 'In Progress');

db_host: 'localhost'
db_port: 5432
db_name: 'postgres'
db_user: 'postgres'
db_password: 'postgres'
required_fields:
  - task_id
  - description
  - status
valid_statuses:
  - Completed
  - Pending
  - In Progress

# File: de-onboarding/utils.py
from typing import Any, Dict  # Import types for annotations
import yaml  # Import PyYAML for parsing YAML configs
from pydantic import BaseModel, ValidationError  # Import Pydantic for validation

def load_yaml(file_path: str) -> Dict[str, Any]:
    """Load YAML file into a dictionary."""
    with open(file_path, "r") as file:  # Open file in read mode
        return yaml.safe_load(file)  # Parse YAML content safely

class ConfigModel(BaseModel):
    """Pydantic model to validate database configuration."""
    db_host: str  # Database host address
    db_port: int  # Database port number
    db_name: str  # Database name
    db_user: str  # Database username
    db_password: str  # Database password
    required_fields: list[str]  # Required task table fields
    valid_statuses: list[str]  # Valid task statuses

def validate_config(config_path: str) -> ConfigModel:
    """Validate YAML config using Pydantic model."""
    config = load_yaml(config_path)  # Load YAML config
    # Pydantic ensures type safety at runtime, preventing errors (Chapter 7)
    try:
        return ConfigModel(**config)  # Validate and return Pydantic model
    except ValidationError as e:
        print(f"Config validation error: {e}")  # Log validation errors
        raise  # Re-raise exception for handling

# File: de-onboarding/task_optimizer.py
from typing import Any, Dict, List  # Import types for annotations
import psycopg2  # Import psycopg2 for PostgreSQL connectivity
from psycopg2.extras import DictCursor  # Import DictCursor for dictionary-like rows
import sqlite3  # Import sqlite3 for reading tasks.db
import json  # Import json for exporting reports
import time  # Import time for measuring execution
from utils import validate_config  # Import config validation function
from pydantic import BaseModel  # Import Pydantic for data modeling

class PerformanceMetrics(BaseModel):
    """Pydantic model to store query performance metrics."""
    query: str  # SQL query text
    plan: List[Dict[str, Any]]  # Query execution plan
    execution_time: float  # Query execution time in seconds
    uses_index: bool  # Whether query uses an index

def seed_postgres(config: BaseModel) -> None:
    """Seed PostgreSQL tasks table from SQLite tasks.db."""
    # Connect to SQLite database to read tasks
    sqlite_conn = sqlite3.connect("data/tasks.db")  # Open tasks.db
    sqlite_cursor = sqlite_conn.cursor()  # Create SQLite cursor
    sqlite_cursor.execute("SELECT task_id, description, status FROM tasks")  # Fetch all tasks
    tasks = sqlite_cursor.fetchall()  # Store tasks as list of tuples
    sqlite_conn.close()  # Close SQLite connection

    # Connect to PostgreSQL database
    conn: psycopg2.connection = psycopg2.connect(
        dbname=config.db_name,  # Database name from config
        user=config.db_user,  # Username
        password=config.db_password,  # Password
        host=config.db_host,  # Host address
        port=config.db_port  # Port number
    )
    cursor: DictCursor = conn.cursor(cursor_factory=DictCursor)  # Create cursor for dictionary results

    # Create tasks table if it doesn't exist
    cursor.execute("""
    CREATE TABLE IF NOT EXISTS tasks (
        task_id TEXT PRIMARY KEY,  -- Unique task identifier
        description TEXT,  -- Task description
        status TEXT  -- Task status
    )
    """)

    # Insert tasks into PostgreSQL, ignoring duplicates
    cursor.executemany(
        "INSERT INTO tasks (task_id, description, status) VALUES (%s, %s, %s) ON CONFLICT DO NOTHING",
        tasks  # List of task tuples
    )
    conn.commit()  # Commit task insertions

    # Create B-tree index on status for faster filtering
    cursor.execute("CREATE INDEX IF NOT EXISTS idx_status ON tasks (status)")
    # Create composite index on status and task_id for multi-condition queries
    cursor.execute("CREATE INDEX IF NOT EXISTS idx_status_task_id ON tasks (status, task_id)")
    conn.commit()  # Commit index creations

    cursor.close()  # Close cursor
    conn.close()  # Close connection
    print("Seeded PostgreSQL tasks table")  # Confirm seeding

def run_query(config: BaseModel, query: str, params: tuple) -> tuple[List[Dict[str, Any]], List[Dict[str, Any]], float]:
    """Run query and return results, plan, and execution time."""
    # Connect to PostgreSQL database
    conn: psycopg2.connection = psycopg2.connect(
        dbname=config.db_name,  # Database name
        user=config.db_user,  # Username
        password=config.db_password,  # Password
        host=config.db_host,  # Host address
        port=config.db_port  # Port number
    )
    cursor: DictCursor = conn.cursor(cursor_factory=DictCursor)  # Create cursor for dictionary results

    # Get query execution plan using EXPLAIN
    cursor.execute(f"EXPLAIN {query}", params)  # Prefix query with EXPLAIN
    plan: List[Dict[str, Any]] = [dict(row) for row in cursor.fetchall()]  # Convert plan to dictionaries

    # Measure query execution time
    start_time = time.time()  # Record start time
    cursor.execute(query, params)  # Execute query
    results: List[Dict[str, Any]] = [dict(row) for row in cursor.fetchall()]  # Convert results to dictionaries
    execution_time = time.time() - start_time  # Calculate duration

    cursor.close()  # Close cursor
    conn.close()  # Close connection
    return results, plan, execution_time  # Return results, plan, and time

def optimize_queries(config: BaseModel) -> Dict[str, List[PerformanceMetrics]]:
    """Run and optimize queries, returning performance metrics."""
    metrics = {"before": [], "after": []}  # Initialize metrics dictionary

    # Run inefficient query (uses Seq Scan due to UPPER)
    query_before = "SELECT * FROM tasks WHERE UPPER(status) = %s"
    results_before, plan_before, time_before = run_query(config, query_before, ("COMPLETED",))
    metrics["before"].append(PerformanceMetrics(  # Store metrics
        query=query_before,  # Query text
        plan=plan_before,  # Execution plan
        execution_time=time_before,  # Execution time
        uses_index="Index Scan" not in str(plan_before)  # Check for index usage
    ))

    # Run optimized query (uses Index Scan)
    query_after = "SELECT * FROM tasks WHERE status = %s"
    results_after, plan_after, time_after = run_query(config, query_after, ("Completed",))
    metrics["after"].append(PerformanceMetrics(  # Store metrics
        query=query_after,  # Query text
        plan=plan_after,  # Execution plan
        execution_time=time_after,  # Execution time
        uses_index="Index Scan" in str(plan_after)  # Check for index usage
    ))

    # Display performance comparison
    print(f"Before Optimization: {time_before:.4f}s, Plan: {plan_before}")
    print(f"After Optimization: {time_after:.4f}s, Plan: {plan_after}")
    return metrics  # Return metrics dictionary

def export_report(metrics: Dict[str, List[PerformanceMetrics]], output_path: str) -> None:
    """Export performance metrics to JSON file."""
    with open(output_path, "w") as file:  # Open file in write mode
        json.dump({k: [m.dict() for m in v] for k, v in metrics.items()}, file, indent=2)  # Write JSON with indentation
    print(f"Exported report to {output_path}")  # Confirm export

def main() -> None:
    """Main function to optimize task database."""
    config_path = "data/config.yaml"  # Path to configuration file
    output_path = "data/performance_report.json"  # Path for performance report

    config = validate_config(config_path)  # Validate and load config
    seed_postgres(config)  # Seed PostgreSQL database
    metrics = optimize_queries(config)  # Optimize queries and collect metrics
    export_report(metrics, output_path)  # Export metrics to JSON

if __name__ == "__main__":
    main()  # Run main function

# File: de-onboarding/tests/test_task_optimizer.py
# Test suite for task_optimizer.py, covering database seeding, query optimization,
# correctness, large datasets, Sharia-compliant audits, index creation failures,
# and empty datasets, ensuring robust testing per Chapter 9's principles.
from typing import Any, Dict, List  # Import types for annotations
import pytest  # Import pytest for testing
import psycopg2  # Import psycopg2 for PostgreSQL connectivity
from psycopg2.extras import DictCursor  # Import DictCursor for dictionary-like rows
from task_optimizer import seed_postgres, run_query, optimize_queries, validate_config  # Import project functions
from utils import ConfigModel  # Import ConfigModel for validation

@pytest.fixture
def config() -> ConfigModel:
    """Fixture to provide validated configuration."""
    return validate_config("data/config.yaml")  # Load and validate config.yaml

@pytest.fixture
def db_connection(config: ConfigModel) -> psycopg2.connection:
    """Fixture to provide PostgreSQL connection."""
    conn = psycopg2.connect(  # Establish connection
        dbname=config.db_name,  # Database name
        user=config.db_user,  # Username
        password=config.db_password,  # Password
        host=config.db_host,  # Host address
        port=config.db_port  # Port number
    )
    yield conn  # Provide connection to tests
    conn.close()  # Close connection after tests

def test_seed_postgres(config: ConfigModel, db_connection: psycopg2.connection) -> None:
    """Test database seeding with tasks from tasks.db."""
    seed_postgres(config)  # Seed PostgreSQL with tasks
    cursor: DictCursor = db_connection.cursor(cursor_factory=DictCursor)  # Create cursor
    cursor.execute("SELECT COUNT(*) FROM tasks")  # Count rows in tasks table
    count: int = cursor.fetchone()[0]  # Get row count
    assert count >= 3, "Tasks table should have at least 3 rows"  # Verify minimum rows
    cursor.close()  # Close cursor

def test_query_optimization(config: ConfigModel) -> None:
    """Test that query optimization improves performance."""
    metrics = optimize_queries(config)  # Run and optimize queries
    assert len(metrics["before"]) == 1, "Should have one unoptimized query"  # Verify unoptimized query count
    assert len(metrics["after"]) == 1, "Should have one optimized query"  # Verify optimized query count
    assert metrics["after"][0].uses_index, "Optimized query should use index"  # Verify index usage
    assert metrics["before"][0].execution_time >= metrics["after"][0].execution_time, "Optimized query should be faster"  # Verify performance improvement

def test_query_correctness(config: ConfigModel) -> None:
    """Test that query returns correct results."""
    query = "SELECT * FROM tasks WHERE status = %s"  # Query for Completed tasks
    results, _, _ = run_query(config, query, ("Completed",))  # Execute query
    assert len(results) == 1, "Should return one Completed task"  # Verify result count
    assert results[0]["status"] == "Completed", "Task status should be Completed"  # Verify status

def test_large_dataset(config: ConfigModel, db_connection: psycopg2.connection) -> None:
    """Test seeding with larger dataset (from Exercise 6)."""
    cursor: DictCursor = db_connection.cursor(cursor_factory=DictCursor)  # Create cursor
    cursor.execute("SELECT COUNT(*) FROM tasks")  # Count initial rows
    initial_count: int = cursor.fetchone()[0]  # Store initial count

    # Simulate large dataset seeding (1000 tasks for test)
    tasks = [(f"T{i:04d}", f"Task {i} for Halal products", "Pending") for i in range(4, 1004)]  # Generate tasks
    cursor.executemany(
        "INSERT INTO tasks (task_id, description, status) VALUES (%s, %s, %s) ON CONFLICT DO NOTHING",
        tasks  # Insert tasks
    )
    db_connection.commit()  # Commit insertions

    cursor.execute("SELECT COUNT(*) FROM tasks")  # Count new rows
    new_count: int = cursor.fetchone()[0]  # Store new count
    assert new_count >= initial_count + len(tasks), "Should add new tasks"  # Verify task addition
    cursor.close()  # Close cursor

def test_audit_query(config: ConfigModel) -> None:
    """Test Sharia-compliant audit query (from Exercise 3)."""
    query = "SELECT * FROM tasks WHERE description LIKE %s"  # Query for Halal-related tasks
    results, _, _ = run_query(config, query, ("Halal%",))  # Execute query
    print(f"Audit query results: {results}")  # Debug: display results
    assert len(results) >= 1, "Should return at least one Halal-related task"  # Verify result count
    assert any("Halal" in r["description"] for r in results), "Tasks should include Halal-related descriptions"  # Verify content

def test_index_creation_failure(config: ConfigModel) -> None:
    """Test index creation on non-existent table raises error."""
    # Connect to PostgreSQL with config
    conn: psycopg2.connection = psycopg2.connect(
        dbname=config.db_name,  # Database name
        user=config.db_user,  # Username
        password=config.db_password,  # Password
        host=config.db_host,  # Host address
        port=config.db_port  # Port number
    )
    cursor: DictCursor = conn.cursor(cursor_factory=DictCursor)  # Create cursor

    # Attempt to create index on non-existent table
    with pytest.raises(psycopg2.errors.UndefinedTable):  # Expect ProgrammingError
        cursor.execute("CREATE INDEX idx_invalid ON nonexistent_table (column)")  # Invalid query
        conn.commit()  # Attempt to commit
    print("Index creation failed as expected")  # Debug: confirm failure

    cursor.close()  # Close cursor
    conn.close()  # Close connection

def test_empty_dataset(config: ConfigModel, db_connection: psycopg2.connection) -> None:
    """Test query behavior on empty tasks table."""
    cursor: DictCursor = db_connection.cursor(cursor_factory=DictCursor)  # Create cursor
    cursor.execute("TRUNCATE TABLE tasks")  # Clear tasks table
    db_connection.commit()  # Commit truncation

    query = "SELECT * FROM tasks WHERE status = %s"  # Query for Completed tasks
    results, _, _ = run_query(config, query, ("Completed",))  # Execute query
    print(f"Empty dataset query results: {results}")  # Debug: display results
    assert len(results) == 0, "Should return zero rows for empty table"  # Verify empty result

    cursor.close()  # Close cursor

{
  "before": [
    {
      "query": "SELECT * FROM tasks WHERE UPPER(status) = %s",
      "plan": [{ "QUERY PLAN": "Seq Scan on tasks ..." }],
      "execution_time": 0.0021,
      "uses_index": false
    }
  ],
  "after": [
    {
      "query": "SELECT * FROM tasks WHERE status = %s",
      "plan": [{ "QUERY PLAN": "Index Scan using idx_status on tasks ..." }],
      "execution_time": 0.0009,
      "uses_index": true
    }
  ]
}

query = "SELECT * FROM tasks WHERE UPPER(description) LIKE %s"

cursor.execute("SELECT * FROM tasks WHERE UPPER(status) = %s", ("COMPLETED",))

from typing import Any  # Import Any for flexible type annotations
import psycopg2  # Import psycopg2 for PostgreSQL connectivity
from psycopg2.extras import DictCursor  # Import DictCursor for dictionary-like rows

def create_description_index(config: Dict[str, Any]) -> None:
    """Create B-tree index on description and estimate size for Halal product tasks."""
    # Connect to PostgreSQL with provided config
    conn: psycopg2.connection = psycopg2.connect(**config)
    cursor: DictCursor = conn.cursor(cursor_factory=DictCursor)  # Create cursor
    # Create B-tree index on description column to speed up Halal task filtering
    cursor.execute("CREATE INDEX IF NOT EXISTS idx_description ON tasks (description)")
    conn.commit()  # Commit index creation

    # Estimate index size using pg_relation_size
    cursor.execute("SELECT pg_relation_size('idx_description')")
    size: int = cursor.fetchone()[0]  # Get size in bytes
    print("Index idx_description created for Halal product tasks")  # Confirm index creation
    print(f"Index Size: {size} bytes")  # Display index size

    cursor.close()  # Close cursor
    conn.close()  # Close connection

# Test configuration
config = {
    "dbname": "postgres",  # Database name
    "user": "postgres",  # Username
    "password": "postgres",  # Password
    "host": "localhost",  # Host address
    "port": "5432"  # Port number
}
create_description_index(config)  # Run function

from typing import Any, List  # Import types for annotations
import psycopg2  # Import psycopg2 for PostgreSQL connectivity
from psycopg2.extras import DictCursor  # Import DictCursor for dictionary-like rows

def analyze_query_plan(config: Dict[str, Any], status: str) -> tuple[List[Dict[str, Any]], List[Dict[str, Any]]]:
    """Analyze query plan with EXPLAIN and EXPLAIN ANALYZE."""
    # Connect to PostgreSQL with config
    conn: psycopg2.connection = psycopg2.connect(**config)
    cursor: DictCursor = conn.cursor(cursor_factory=DictCursor)  # Create cursor

    # Run EXPLAIN to get estimated query plan
    cursor.execute("EXPLAIN SELECT * FROM tasks WHERE status = %s", (status,))
    explain_plan: List[Dict[str, Any]] = [dict(row) for row in cursor.fetchall()]  # Convert plan to dictionaries

    # Run EXPLAIN ANALYZE to get actual execution details (optional)
    cursor.execute("EXPLAIN ANALYZE SELECT * FROM tasks WHERE status = %s", (status,))
    analyze_plan: List[Dict[str, Any]] = [dict(row) for row in cursor.fetchall()]  # Convert plan to dictionaries

    cursor.close()  # Close cursor
    conn.close()  # Close connection
    print("EXPLAIN Plan:", explain_plan)  # Display estimated plan
    print("EXPLAIN ANALYZE Plan:", analyze_plan)  # Display actual plan
    return explain_plan, analyze_plan  # Return both plans

# Test configuration
config = {
    "dbname": "postgres",  # Database name
    "user": "postgres",  # Username
    "password": "postgres",  # Password
    "host": "localhost",  # Host address
    "port": "5432"  # Port number
}
analyze_query_plan(config, "Completed")  # Run function

from typing import Any, List  # Import types for annotations
import psycopg2  # Import psycopg2 for PostgreSQL connectivity
from psycopg2.extras import DictCursor  # Import DictCursor for dictionary-like rows

def optimize_query(config: Dict[str, Any]) -> List[Dict[str, Any]]:
    """Optimize query for Halal product audit to use index."""
    # Connect to PostgreSQL with config
    conn: psycopg2.connection = psycopg2.connect(**config)
    cursor: DictCursor = conn.cursor(cursor_factory=DictCursor)  # Create cursor
    optimized_query = "SELECT * FROM tasks WHERE description LIKE %s"  # Optimized query for Halal tasks
    # Run EXPLAIN to analyze query plan
    cursor.execute("EXPLAIN " + optimized_query, ("Halal%",))
    plan: List[Dict[str, Any]] = [dict(row) for row in cursor.fetchall()]  # Convert plan to dictionaries
    cursor.close()  # Close cursor
    conn.close()  # Close connection
    print("Optimized Query:", optimized_query)  # Display query
    print("Plan:", plan)  # Display plan
    return plan  # Return query plan

# Test configuration
config = {
    "dbname": "postgres",  # Database name
    "user": "postgres",  # Username
    "password": "postgres",  # Password
    "host": "localhost",  # Host address
    "port": "5432"  # Port number
}
optimize_query(config)  # Run function

from typing import Any, List  # Import types for annotations
import psycopg2  # Import psycopg2 for PostgreSQL connectivity
from psycopg2.extras import DictCursor  # Import DictCursor for dictionary-like rows
import time  # Import time for measuring execution

def measure_query_time(config: Dict[str, Any], status: str) -> float:
    """Measure query execution time."""
    # Connect to PostgreSQL with config
    conn: psycopg2.connection = psycopg2.connect(**config)
    cursor: DictCursor = conn.cursor(cursor_factory=DictCursor)  # Create cursor
    start_time = time.time()  # Record start time
    # Execute query to fetch tasks by status
    cursor.execute("SELECT * FROM tasks WHERE status = %s", (status,))
    results: List[Dict[str, Any]] = [dict(row) for row in cursor.fetchall()]  # Convert results to dictionaries
    execution_time = time.time() - start_time  # Calculate duration
    cursor.close()  # Close cursor
    conn.close()  # Close connection
    print(f"Query Time: {execution_time:.4f} seconds")  # Display execution time
    return execution_time  # Return execution time

# Test configuration
config = {
    "dbname": "postgres",  # Database name
    "user": "postgres",  # Username
    "password": "postgres",  # Password
    "host": "localhost",  # Host address
    "port": "5432"  # Port number
}
measure_query_time(config, "Completed")  # Run function

from typing import Any, List  # Import types for annotations
import psycopg2  # Import psycopg2 for PostgreSQL connectivity
from psycopg2.extras import DictCursor  # Import DictCursor for dictionary-like rows

def debug_slow_query(config: Dict[str, Any]) -> List[Dict[str, Any]]:
    """Fix slow query and update statistics."""
    # Connect to PostgreSQL with config
    conn: psycopg2.connection = psycopg2.connect(**config)
    cursor: DictCursor = conn.cursor(cursor_factory=DictCursor)  # Create cursor

    # Update table statistics to improve query planning
    cursor.execute("ANALYZE tasks")
    print("Statistics Updated: ANALYZE completed")  # Confirm statistics update

    # Analyze optimized query plan
    cursor.execute("EXPLAIN SELECT * FROM tasks WHERE status = %s", ("Completed",))
    plan: List[Dict[str, Any]] = [dict(row) for row in cursor.fetchall()]  # Convert plan to dictionaries

    cursor.close()  # Close cursor
    conn.close()  # Close connection
    print("Optimized Plan:", plan)  # Display plan
    return plan  # Return query plan

# Test configuration
config = {
    "dbname": "postgres",  # Database name
    "user": "postgres",  # Username
    "password": "postgres",  # Password
    "host": "localhost",  # Host address
    "port": "5432"  # Port number
}
debug_slow_query(config)  # Run function

from typing import Any, List  # Import types for annotations
import psycopg2  # Import psycopg2 for PostgreSQL connectivity
from psycopg2.extras import DictCursor  # Import DictCursor for dictionary-like rows
import time  # Import time for measuring execution
from random import choice  # Import choice for random status selection

def generate_large_dataset(config: Dict[str, Any], num_rows: int = 10000) -> float:
    """Generate large dataset and measure query performance for Hijra Group’s task management."""
    statuses = ["Completed", "Pending", "In Progress"]  # Valid task statuses
    # Generate tasks with unique IDs and Halal-related descriptions
    tasks = [
        (f"T{i:04d}", f"Task {i} for Halal products", choice(statuses))
        for i in range(4, num_rows + 4)
    ]

    # Connect to PostgreSQL with config
    conn: psycopg2.connection = psycopg2.connect(**config)
    cursor: DictCursor = conn.cursor(cursor_factory=DictCursor)  # Create cursor

    # Insert tasks, ignoring duplicates
    cursor.executemany(
        "INSERT INTO tasks (task_id, description, status) VALUES (%s, %s, %s) ON CONFLICT DO NOTHING",
        tasks
    )
    conn.commit()  # Commit insertions
    print(f"Generated {num_rows} tasks for Hijra Group’s task management")  # Confirm task generation

    # Measure query performance on large dataset
    start_time = time.time()  # Record start time
    cursor.execute("SELECT * FROM tasks WHERE status = %s", ("Pending",))  # Query tasks by status
    results: List[Dict[str, Any]] = [dict(row) for row in cursor.fetchall()]  # Convert results to dictionaries
    execution_time = time.time() - start_time  # Calculate duration

    cursor.close()  # Close cursor
    conn.close()  # Close connection
    print(f"Query Time: {execution_time:.4f} seconds")  # Display execution time
    return execution_time  # Return execution time

# Test configuration
config = {
    "dbname": "postgres",  # Database name
    "user": "postgres",  # Username
    "password": "postgres",  # Password
    "host": "localhost",  # Host address
    "port": "5432"  # Port number
}
generate_large_dataset(config)  # Run function

def explain_tradeoffs(output_path: str) -> None:
    """Explain indexing trade-offs and save to file."""
    # Define explanation for indexing trade-offs in Hijra Group’s context
    explanation = (
        "Indexing in PostgreSQL speeds up queries (e.g., filtering millions of tasks by status) with O(log n) lookup time "
        "but increases storage and slows INSERT/UPDATE operations due to index maintenance (O(log n)). "
        "For Hijra Group’s task management, indexes on status enable fast analytics for Sharia-compliant product tasks "
        "across large transaction volumes, but frequent task updates require careful index selection to balance performance."
    )
    # Write explanation to file
    with open(output_path, "w") as file:  # Open file in write mode
        file.write(explanation)  # Write explanation
    print(f"Saved explanation to {output_path}")  # Confirm file creation

# Test
explain_tradeoffs("data/ex7_tradeoffs.txt")  # Run function