22 - PostgreSQL Indexing and Optimization

# File: de-onboarding/index_basics.py
from typing import Any  # For type annotations
import psycopg2  # For PostgreSQL connection

# Connect to database
conn: psycopg2.connection = psycopg2.connect(
    dbname="sales_db",
    user="postgres",
    password="password",
    host="localhost",
    port="5432"
)
cursor: psycopg2.cursor = conn.cursor()

# Create transactions table with JSONB column
cursor.execute("""
    CREATE TABLE IF NOT EXISTS transactions (
        transaction_id TEXT PRIMARY KEY,
        product TEXT,
        price REAL,
        quantity INTEGER,
        date DATE,
        metadata JSONB
    )
""")

# Insert sample data
cursor.execute("""
    INSERT INTO transactions (transaction_id, product, price, quantity, date, metadata)
    VALUES
        ('T001', 'Halal Laptop', 999.99, 2, '2023-10-01', '{"category": "electronics", "details": {"subcategory": "laptop"}}'),
        ('T002', 'Halal Mouse', 24.99, 10, '2023-10-02', '{"category": "accessories", "details": {"subcategory": "mouse"}}'),
        ('T003', 'Halal Keyboard', 49.99, 5, '2023-10-03', '{"category": "accessories", "details": {"subcategory": "keyboard"}}')
""")

# Create B-tree index
cursor.execute("CREATE INDEX IF NOT EXISTS idx_transaction_id ON transactions (transaction_id)")
conn.commit()

# Query with index
cursor.execute("SELECT * FROM transactions WHERE transaction_id = 'T001'")
result: list[Any] = cursor.fetchall()
print("Query Result:", result)  # Debug: print result

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

# Expected Output:
# Query Result: [('T001', 'Halal Laptop', 999.99, 2, datetime.date(2023, 10, 1), {'category': 'electronics', 'details': {'subcategory': 'laptop'}})]

# File: de-onboarding/explain_query.py
from typing import Any
import psycopg2

conn: psycopg2.connection = psycopg2.connect(
    dbname="sales_db",
    user="postgres",
    password="password",
    host="localhost",
    port="5432"
)
cursor: psycopg2.cursor = conn.cursor()

# Analyze query plan
cursor.execute("EXPLAIN SELECT * FROM transactions WHERE transaction_id = 'T001'")
plan: list[Any] = cursor.fetchall()
print("Query Plan:")  # Debug
for row in plan:
    print(row[0])  # Print plan steps

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

# Expected Output (abridged):
# Query Plan:
# Index Scan using idx_transaction_id on transactions  (cost=0.15..8.17 rows=1 width=64)

# File: de-onboarding/gin_index.py
from typing import Any
import psycopg2

conn: psycopg2.connection = psycopg2.connect(
    dbname="sales_db",
    user="postgres",
    password="password",
    host="localhost",
    port="5432"
)
cursor: psycopg2.cursor = conn.cursor()

# Create GIN index
cursor.execute("CREATE INDEX IF NOT EXISTS idx_metadata ON transactions USING GIN (metadata)")
conn.commit()

# Simple JSONB query
cursor.execute("SELECT * FROM transactions WHERE metadata @> '{\"category\": \"electronics\"}'")
simple_result: list[Any] = cursor.fetchall()
print("Simple JSONB Query Result:", simple_result)  # Debug: print result

# Nested JSONB query
cursor.execute("SELECT * FROM transactions WHERE metadata->'details'->>'subcategory' = 'laptop'")
nested_result: list[Any] = cursor.fetchall()
print("Nested JSONB Query Result:", nested_result)  # Debug: print result

# Analyze plan for nested query
cursor.execute("EXPLAIN SELECT * FROM transactions WHERE metadata->'details'->>'subcategory' = 'laptop'")
plan: list[Any] = cursor.fetchall()
print("Nested Query Plan:")  # Debug
for row in plan:
    print(row[0])
print("Note: Nested JSONB queries with ->> may use sequential scans, as GIN indexes optimize @> queries. Expression indexes (Chapter 48) can optimize such queries.")

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

# Expected Output (abridged):
# Simple JSONB Query Result: [('T001', 'Halal Laptop', 999.99, 2, datetime.date(2023, 10, 1), {'category': 'electronics', 'details': {'subcategory': 'laptop'}})]
# Nested JSONB Query Result: [('T001', 'Halal Laptop', 999.99, 2, datetime.date(2023, 10, 1), {'category': 'electronics', 'details': {'subcategory': 'laptop'}})]
# Nested Query Plan:
# Seq Scan on transactions  (cost=0.00..1.03 rows=1 width=64)
#   Filter: ((metadata -> 'details' ->> 'subcategory'::text) = 'laptop'::text)
# Note: Nested JSONB queries with ->> may use sequential scans, as GIN indexes optimize @> queries. Expression indexes (Chapter 48) can optimize such queries.

# File: de-onboarding/range_query.py
from typing import Any
import psycopg2

conn: psycopg2.connection = psycopg2.connect(
    dbname="sales_db",
    user="postgres",
    password="password",
    host="localhost",
    port="5432"
)
cursor: psycopg2.cursor = conn.cursor()

# Create index on date
cursor.execute("CREATE INDEX IF NOT EXISTS idx_date ON transactions (date)")

# Optimized range query
cursor.execute("""
    SELECT product, SUM(price * quantity) as total_sales
    FROM transactions
    WHERE date BETWEEN '2023-10-01' AND '2023-10-03'
    GROUP BY product
""")
results: list[Any] = cursor.fetchall()
print("Range Query Results:", results)  # Debug

# Analyze plan
cursor.execute("""
    EXPLAIN SELECT product, SUM(price * quantity) as total_sales
    FROM transactions
    WHERE date BETWEEN '2023-10-01' AND '2023-10-03'
    GROUP BY product
""")
plan: list[Any] = cursor.fetchall()
print("Query Plan:")  # Debug
for row in plan:
    print(row[0])

conn.commit()
cursor.close()
conn.close()

# Expected Output (abridged):
# Range Query Results: [('Halal Laptop', 1999.98), ('Halal Mouse', 249.9), ('Halal Keyboard', 249.95)]
# Query Plan:
# GroupAggregate  (cost=0.15..8.45 rows=3 width=36)
#   ->  Index Scan using idx_date on transactions  (cost=0.15..8.27 rows=3 width=28)

# File: de-onboarding/maintenance.py
from typing import Any
import psycopg2

conn: psycopg2.connection = psycopg2.connect(
    dbname="sales_db",
    user="postgres",
    password="password",
    host="localhost",
    port="5432"
)
conn.set_session(autocommit=True)  # Enable autocommit for VACUUM
cursor: psycopg2.cursor = conn.cursor()

# Vacuum and analyze
cursor.execute("VACUUM ANALYZE transactions")
print("Maintenance completed")  # Confirm

cursor.close()
conn.close()

# Expected Output:
# Maintenance completed

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

# File: de-onboarding/seed_transactions.py
import pandas as pd
import random
from datetime import datetime, timedelta
import json

def generate_transactions(n: int = 1000) -> None:
    """Generate transactions.csv with 1000 rows."""
    products = ["Halal Laptop", "Halal Mouse", "Halal Keyboard", "Monitor", ""]
    categories = ["electronics", "accessories", "misc"]
    subcategories = ["laptop", "mouse", "keyboard", "monitor", "other"]
    start_date = datetime(2023, 10, 1)

    data = {
        "transaction_id": [f"T{i:03d}" for i in range(1, n + 1)],
        "product": [random.choice(products) for _ in range(n)],
        "price": [round(random.uniform(5.0, 1000.0), 2) for _ in range(n)],
        "quantity": [random.randint(1, 150) for _ in range(n)],
        "date": [(start_date + timedelta(days=random.randint(0, 30))).strftime("%Y-%m-%d") for _ in range(n)],
        "metadata": [json.dumps({
            "category": random.choice(categories),
            "details": {"subcategory": random.choice(subcategories)}
        }) for _ in range(n)]
    }

    df = pd.DataFrame(data)
    df.to_csv("data/transactions.csv", index=False)
    print(f"Generated data/transactions.csv with {n} rows")

if __name__ == "__main__":
    generate_transactions()

# File: de-onboarding/utils.py (updated from Chapter 17)
from typing import Any, Dict
from pydantic import BaseModel, ValidationError
import json
import re

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

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(".")
    if len(parts) != 2 or not parts[0].isdigit() or not parts[1].isdigit():
        return False
    return len(parts[1]) <= max_decimals

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

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

def apply_valid_decimals(x: Any, max_decimals: int) -> bool:
    """Apply decimal validation."""
    return is_numeric(str(x), max_decimals)

def is_valid_date(s: str) -> bool:
    """Check if string is a valid date in YYYY-MM-DD format."""
    try:
        from datetime import datetime
        datetime.strptime(s, "%Y-%m-%d")
        return True
    except ValueError:
        return False

def parse_explain_cost(plan: list[Any]) -> float:
    """Parse total cost from EXPLAIN output."""
    for line in plan:
        match = re.search(r"cost=[\d.]+..([\d.]+)", line[0])
        if match:
            return float(match.group(1))
    return 0.0

class Transaction(BaseModel):
    """Pydantic model for transaction validation."""
    transaction_id: str
    product: str
    price: float
    quantity: int
    date: str
    metadata: dict

def validate_transaction(row: Dict[str, Any], config: Dict[str, Any]) -> bool:
    """Validate transaction using Pydantic and config rules."""
    try:
        # Clean data
        metadata = row["metadata"]
        if isinstance(metadata, str):
            metadata = json.loads(metadata)
        cleaned = {
            "transaction_id": clean_string(str(row["transaction_id"])),
            "product": clean_string(str(row["product"])),
            "price": float(row["price"]),
            "quantity": int(row["quantity"]),
            "date": str(row["date"]),
            "metadata": metadata
        }
        # Validate with Pydantic
        Transaction(**cleaned)
        # Config-based validation
        if not cleaned["product"].startswith(config["product_prefix"]):
            print(f"Invalid transaction: product lacks '{config['product_prefix']}' prefix: {row}")
            return False
        if cleaned["price"] < config["min_price"] or cleaned["price"] <= 0:
            print(f"Invalid transaction: invalid price: {row}")
            return False
        if cleaned["quantity"] > config["max_quantity"]:
            print(f"Invalid transaction: invalid quantity: {row}")
            return False
        if not apply_valid_decimals(cleaned["price"], config["max_decimals"]):
            print(f"Invalid transaction: too many decimals: {row}")
            return False
        if not is_valid_date(cleaned["date"]):
            print(f"Invalid transaction: invalid date: {row}")
            return False
        print(f"Valid transaction: {row}")
        return True
    except (ValidationError, ValueError, json.JSONDecodeError) as e:
        print(f"Invalid transaction: {e}: {row}")
        return False

# File: de-onboarding/optimized_processor.py
from typing import Any, Dict, List, Tuple
import pandas as pd
import psycopg2
import yaml
import json
import os
import time
from pydantic import BaseModel
import utils

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

def setup_database(conn: psycopg2.connection) -> None:
    """Create transactions table."""
    cursor = conn.cursor()
    cursor.execute("""
        CREATE TABLE IF NOT EXISTS transactions (
            transaction_id TEXT PRIMARY KEY,
            product TEXT,
            price REAL,
            quantity INTEGER,
            date DATE,
            metadata JSONB
        )
    """)
    cursor.execute("TRUNCATE TABLE transactions")  # Clear table for testing
    conn.commit()
    cursor.close()

def load_and_insert_transactions(csv_path: str, config: Dict[str, Any], conn: psycopg2.connection) -> Tuple[pd.DataFrame, int, int]:
    """Load and insert valid transactions."""
    print(f"Loading CSV: {csv_path}")
    df = pd.read_csv(csv_path)
    print("Initial DataFrame:")
    print(df.head())

    # Validate required fields
    required_fields = config["required_fields"]
    missing_fields = [f for f in required_fields if f not in df.columns]
    if missing_fields:
        print(f"Missing columns: {missing_fields}")
        return pd.DataFrame(), 0, len(df)

    # Check for duplicates
    if df["transaction_id"].duplicated().any():
        print("Duplicate transaction IDs found")
        df = df.drop_duplicates(subset="transaction_id", keep="first")

    # Filter valid rows
    df = df.dropna(subset=["product", "price", "quantity", "date", "metadata"])
    df = df[df.apply(lambda row: utils.validate_transaction(row, config), axis=1)]
    print("Validated DataFrame:")
    print(df.head())

    # Insert into PostgreSQL
    cursor = conn.cursor()
    for _, row in df.iterrows():
        try:
            metadata = row["metadata"]
            if isinstance(metadata, str):
                metadata = json.loads(metadata)
            cursor.execute("""
                INSERT INTO transactions (transaction_id, product, price, quantity, date, metadata)
                VALUES (%s, %s, %s, %s, %s, %s)
            """, (row["transaction_id"], row["product"], row["price"], row["quantity"], row["date"], json.dumps(metadata)))
        except psycopg2.IntegrityError as e:
            print(f"Duplicate transaction_id: {row['transaction_id']}")
            conn.rollback()
            continue
    conn.commit()
    cursor.close()

    return df, len(df), len(df)

def create_indexes(conn: psycopg2.connection) -> None:
    """Create B-tree and GIN indexes."""
    cursor = conn.cursor()
    cursor.execute("CREATE INDEX IF NOT EXISTS idx_transaction_id ON transactions (transaction_id)")
    cursor.execute("CREATE INDEX IF NOT EXISTS idx_date ON transactions (date)")
    cursor.execute("CREATE INDEX IF NOT EXISTS idx_metadata ON transactions USING GIN (metadata)")
    conn.commit()
    cursor.close()
    print("Indexes created")

def estimate_index_sizes(conn: psycopg2.connection, row_count: int) -> Dict[str, Dict[str, float]]:
    """Estimate and query actual index sizes in MB."""
    cursor = conn.cursor()
    cursor.execute("""
        SELECT indexname, pg_relation_size(indexname::regclass) / 1048576.0 as size_mb
        FROM pg_indexes
        WHERE tablename = 'transactions'
    """)
    actual_sizes = {row[0]: row[1] for row in cursor.fetchall()}
    cursor.close()

    # Heuristic estimates
    btree_size_mb = (row_count * 10) / 1_000_000  # ~10 bytes per row
    gin_size_mb = (row_count * 15) / 1_000_000    # ~15 bytes per row
    estimated_sizes = {
        "idx_transaction_id": btree_size_mb,
        "idx_date": btree_size_mb,
        "idx_metadata": gin_size_mb
    }

    return {
        "actual": actual_sizes,
        "estimated": estimated_sizes
    }

def format_index_sizes(index_sizes: Dict[str, Dict[str, float]]) -> str:
    """Format index sizes as a table."""
    header = f"{'Index Name':<25} {'Actual Size (MB)':<20} {'Estimated Size (MB)':<20}"
    rows = []
    for index in index_sizes["actual"]:
        actual = index_sizes["actual"].get(index, 0.0)
        estimated = index_sizes["estimated"].get(index, 0.0)
        rows.append(f"{index:<25} {actual:<20.2f} {estimated:<20.2f}")
    return "\n".join([header, "-" * 65] + rows)

def optimize_queries(conn: psycopg2.connection) -> Tuple[List[Any], List[Any], List[Any], List[Any], float, float]:
    """Run optimized queries and analyze plans."""
    cursor = conn.cursor()

    # Sales query
    start_time = time.time()
    cursor.execute("""
        SELECT product, SUM(price * quantity) as total_sales
        FROM transactions
        WHERE date BETWEEN '2023-10-01' AND '2023-10-03'
        GROUP BY product
    """)
    sales_results = cursor.fetchall()
    sales_time = time.time() - start_time
    print(f"Sales Query Results: {sales_results} (Time: {sales_time:.3f}s)")

    cursor.execute("""
        EXPLAIN SELECT product, SUM(price * quantity) as total_sales
        FROM transactions
        WHERE date BETWEEN '2023-10-01' AND '2023-10-03'
        GROUP BY product
    """)
    sales_plan = cursor.fetchall()
    print("Sales Query Plan:")
    for row in sales_plan:
        print(row[0])

    # Nested JSONB query
    start_time = time.time()
    cursor.execute("""
        SELECT product, COUNT(*) as count
        FROM transactions
        WHERE metadata->'details'->>'subcategory' = 'laptop'
        GROUP BY product
    """)
    jsonb_results = cursor.fetchall()
    jsonb_time = time.time() - start_time
    print(f"Nested JSONB Query Results: {jsonb_results} (Time: {jsonb_time:.3f}s)")

    cursor.execute("""
        EXPLAIN SELECT product, COUNT(*) as count
        FROM transactions
        WHERE metadata->'details'->>'subcategory' = 'laptop'
        GROUP BY product
    """)
    jsonb_plan = cursor.fetchall()
    print("Nested JSONB Query Plan:")
    for row in jsonb_plan:
        print(row[0])

    cursor.close()
    return sales_results, sales_plan, jsonb_results, jsonb_plan, sales_time, jsonb_time

def run_maintenance(conn: psycopg2.connection) -> None:
    """Run VACUUM ANALYZE."""
    conn.set_session(autocommit=True)
    cursor = conn.cursor()
    cursor.execute("VACUUM ANALYZE transactions")
    cursor.close()
    print("Maintenance completed")

def export_results(sales_results: List[Any], jsonb_results: List[Any], json_path: str) -> None:
    """Export results to JSON."""
    result_dict = {
        "total_sales_by_product": [
            {"product": row[0], "total_sales": float(row[1])} for row in sales_results
        ],
        "laptop_products": [
            {"product": row[0], "count": int(row[1])} for row in jsonb_results
        ]
    }
    print(f"Writing to: {json_path}")
    with open(json_path, "w") as file:
        json.dump(result_dict, file, indent=2)
    print(f"Exported to {json_path}")

def main() -> None:
    """Main function to optimize transaction database."""
    csv_path = "data/transactions.csv"
    config_path = "data/config.yaml"
    json_path = "data/optimized_results.json"

    # Generate dataset if not exists
    if not os.path.exists(csv_path):
        print("Generating transactions.csv")
        from seed_transactions import generate_transactions
        generate_transactions()

    config = read_config(config_path)
    conn = psycopg2.connect(
        dbname="sales_db",
        user="postgres",
        password="password",
        host="localhost",
        port="5432"
    )

    setup_database(conn)
    df, valid_count, total_count = load_and_insert_transactions(csv_path, config, conn)
    create_indexes(conn)
    index_sizes = estimate_index_sizes(conn, total_count)
    sales_results, sales_plan, jsonb_results, jsonb_plan, sales_time, jsonb_time = optimize_queries(conn)
    run_maintenance(conn)
    export_results(sales_results, jsonb_results, json_path)

    print("\nOptimization Report:")
    print(f"Total Records Processed: {total_count}")
    print(f"Valid Records: {valid_count}")
    print(f"Invalid Records: {total_count - valid_count}")
    print(f"Sales Query Time: {sales_time:.3f}s")
    print(f"Nested JSONB Query Time: {jsonb_time:.3f}s")
    print("Index Sizes:")
    print(format_index_sizes(index_sizes))

    conn.close()

if __name__ == "__main__":
    main()

# File: de-onboarding/test_optimized_processor.py
import pytest
import psycopg2
import pandas as pd
import time
from optimized_processor import setup_database, load_and_insert_transactions, create_indexes, optimize_queries, run_maintenance
from typing import Any

@pytest.fixture
def db_connection():
    """Create a test database connection."""
    conn = psycopg2.connect(
        dbname="sales_db",
        user="postgres",
        password="password",
        host="localhost",
        port="5432"
    )
    yield conn
    conn.close()

def test_pipeline(db_connection: psycopg2.connection) -> None:
    """Test the full optimization pipeline."""
    config = {
        "min_price": 10.0,
        "max_quantity": 100,
        "required_fields": ["transaction_id", "product", "price", "quantity", "date", "metadata"],
        "product_prefix": "Halal",
        "max_decimals": 2
    }

    setup_database(db_connection)
    df, valid_count, total_count = load_and_insert_transactions("data/transactions.csv", config, db_connection)
    assert valid_count > 0, f"Expected valid records, got {valid_count}"

    # Test performance without indexes
    cursor = db_connection.cursor()
    start_time = time.time()
    cursor.execute("""
        SELECT product, SUM(price * quantity) as total_sales
        FROM transactions
        WHERE date BETWEEN '2023-10-01' AND '2023-10-03'
        GROUP BY product
    """)
    no_index_time = time.time() - start_time
    cursor.close()

    create_indexes(db_connection)
    cursor = db_connection.cursor()
    cursor.execute("SELECT indexname FROM pg_indexes WHERE tablename = 'transactions'")
    indexes = [row[0] for row in cursor.fetchall()]
    assert "idx_transaction_id" in indexes, "Transaction ID index missing"
    assert "idx_date" in indexes, "Date index missing"
    assert "idx_metadata" in indexes, "Metadata index missing"

    # Test performance with indexes
    sales_results, sales_plan, jsonb_results, jsonb_plan, sales_time, jsonb_time = optimize_queries(db_connection)
    assert sales_time < no_index_time / 2, f"Indexed sales query ({sales_time:.3f}s) not 2x faster than non-indexed ({no_index_time:.3f}s)"
    assert len(sales_results) > 0, f"Expected sales results, got {len(sales_results)}"
    plan_text = " ".join(row[0] for row in sales_plan)
    assert "Index Scan" in plan_text, "Sales query not using index scan"

    assert len(jsonb_results) > 0, f"Expected JSONB results, got {len(jsonb_results)}"

    # Test duplicate handling
    df_duplicate = pd.DataFrame({
        "transaction_id": ["T001"],
        "product": ["Halal Laptop"],
        "price": [999.99],
        "quantity": [2],
        "date": ["2023-10-01"],
        "metadata": ['{"category": "electronics", "details": {"subcategory": "laptop"}}']
    })
    df_duplicate.to_csv("data/duplicate.csv", index=False)
    _, valid_count, _ = load_and_insert_transactions("data/duplicate.csv", config, db_connection)
    assert valid_count == 0, f"Expected 0 valid records for duplicates, got {valid_count}"

    # Test malformed JSONB and invalid date
    df_invalid = pd.DataFrame({
        "transaction_id": ["T999"],
        "product": ["Halal Laptop"],
        "price": [999.99],
        "quantity": [2],
        "date": ["2023-13-01"],
        "metadata": ['{category: electronics}']
    })
    df_invalid.to_csv("data/invalid.csv", index=False)
    _, valid_count, _ = load_and_insert_transactions("data/invalid.csv", config, db_connection)
    assert valid_count == 0, f"Expected 0 valid records for invalid data, got {valid_count}"

    run_maintenance(db_connection)
    cursor.close()

{
  "total_sales_by_product": [
    { "product": "Halal Laptop", "total_sales": 1999.98 },
    { "product": "Halal Mouse", "total_sales": 249.9 },
    { "product": "Halal Keyboard", "total_sales": 249.95 }
  ],
  "laptop_products": [{ "product": "Halal Laptop", "count": 200 }]
}

cursor.execute("""
    SELECT t.product, t.quantity
    FROM transactions t
    JOIN inventory i ON t.product = i.product
    WHERE EXTRACT(YEAR FROM t.date) = 2023
""")

from typing import Any
import psycopg2

def create_product_index() -> None:
    conn: psycopg2.connection = psycopg2.connect(
        dbname="sales_db", user="postgres", password="password", host="localhost", port="5432"
    )
    cursor: psycopg2.cursor = conn.cursor()
    cursor.execute("CREATE INDEX IF NOT EXISTS idx_product ON transactions (product)")
    conn.commit()
    cursor.close()
    conn.close()
    print("Index idx_product created")

create_product_index()

from typing import Any
import psycopg2

def analyze_product_query() -> None:
    conn: psycopg2.connection = psycopg2.connect(
        dbname="sales_db", user="postgres", password="password", host="localhost", port="5432"
    )
    cursor: psycopg2.cursor = conn.cursor()
    cursor.execute("EXPLAIN SELECT * FROM transactions WHERE product = 'Halal Laptop'")
    plan: list[Any] = cursor.fetchall()
    print("Query Plan:")
    for row in plan:
        print(row[0])
    cursor.close()
    conn.close()

analyze_product_query()

from typing import Any
import psycopg2

def optimize_jsonb_query() -> None:
    conn: psycopg2.connection = psycopg2.connect(
        dbname="sales_db", user="postgres", password="password", host="localhost", port="5432"
    )
    cursor: psycopg2.cursor = conn.cursor()
    cursor.execute("""
        SELECT product, COUNT(*) as count
        FROM transactions
        WHERE metadata->'details'->>'subcategory' = 'laptop'
        GROUP BY product
    """)
    results: list[Any] = cursor.fetchall()
    print("Query Results:", results)
    cursor.execute("""
        EXPLAIN SELECT product, COUNT(*) as count
        FROM transactions
        WHERE metadata->'details'->>'subcategory' = 'laptop'
        GROUP BY product
    """)
    plan: list[Any] = cursor.fetchall()
    print("Query Plan:")
    for row in plan:
        print(row[0])
    cursor.close()
    conn.close()

optimize_jsonb_query()

from typing import Any
import psycopg2

def run_maintenance() -> None:
    conn: psycopg2.connection = psycopg2.connect(
        dbname="sales_db", user="postgres", password="password", host="localhost", port="5432"
    )
    conn.set_session(autocommit=True)
    cursor: psycopg2.cursor = conn.cursor()
    cursor.execute("VACUUM ANALYZE transactions")
    cursor.close()
    conn.close()
    print("Maintenance completed")

run_maintenance()

from typing import Any
import psycopg2

def fix_slow_query() -> None:
    conn: psycopg2.connection = psycopg2.connect(
        dbname="sales_db", user="postgres", password="password", host="localhost", port="5432"
    )
    cursor: psycopg2.cursor = conn.cursor()
    cursor.execute("""
        CREATE TABLE IF NOT EXISTS inventory (
            product TEXT PRIMARY KEY,
            stock INTEGER
        )
    """)
    cursor.execute("INSERT INTO inventory (product, stock) VALUES ('Halal Laptop', 10) ON CONFLICT DO NOTHING")
    cursor.execute("""
        SELECT t.product, t.quantity
        FROM transactions t
        JOIN inventory i ON t.product = i.product
        WHERE t.date >= '2023-01-01' AND t.date < '2024-01-01'
    """)
    results: list[Any] = cursor.fetchall()
    print("Query Results:", results)
    cursor.execute("""
        EXPLAIN SELECT t.product, t.quantity
        FROM transactions t
        JOIN inventory i ON t.product = i.product
        WHERE t.date >= '2023-01-01' AND t.date < '2024-01-01'
    """)
    plan: list[Any] = cursor.fetchall()
    print("Query Plan:")
    for row in plan:
        print(row[0])
    conn.commit()
    cursor.close()
    conn.close()

fix_slow_query()

from typing import Any
import psycopg2

def analyze_index_tradeoffs() -> None:
    conn: psycopg2.connection = psycopg2.connect(
        dbname="sales_db", user="postgres", password="password", host="localhost", port="5432"
    )
    cursor: psycopg2.cursor = conn.cursor()
    cursor.execute("""
        SELECT indexname, pg_size_pretty(pg_relation_size(indexname::regclass)) as size
        FROM pg_indexes
        WHERE tablename = 'transactions'
    """)
    sizes = cursor.fetchall()
    cursor.close()
    conn.close()

    analysis = """
B-tree indexes are ideal for equality and range queries (e.g., transaction_id, date), offering O(log n) lookups but require ~10MB for 1M rows. GIN indexes optimize JSONB searches (e.g., metadata), with ~15–20MB storage and slower updates due to complex structures. Use B-tree for structured columns, GIN for JSONB in Hijra Group’s analytics.
Actual Index Sizes:
"""
    for index, size in sizes:
        analysis += f"  {index}: {size}\n"

    with open("ex6_concepts.txt", "w") as file:
        file.write(analysis.strip())
    print("Analysis saved to ex6_concepts.txt")

analyze_index_tradeoffs()

from typing import Any
import psycopg2

def extend_with_gin_index() -> None:
    conn: psycopg2.connection = psycopg2.connect(
        dbname="sales_db", user="postgres", password="password", host="localhost", port="5432"
    )
    cursor: psycopg2.cursor = conn.cursor()
    cursor.execute("CREATE INDEX IF NOT EXISTS idx_metadata ON transactions USING GIN (metadata)")
    cursor.execute("""
        SELECT product, COUNT(*) as count
        FROM transactions
        WHERE metadata->'details'->>'subcategory' = 'laptop'
        GROUP BY product
    """)
    results: list[Any] = cursor.fetchall()
    print("Extended Query Results:", results)
    cursor.execute("""
        EXPLAIN SELECT product, COUNT(*) as count
        FROM transactions
        WHERE metadata->'details'->>'subcategory' = 'laptop'
        GROUP BY product
    """)
    plan: list[Any] = cursor.fetchall()
    print("Query Plan:")
    for row in plan:
        print(row[0])
    conn.commit()
    cursor.close()
    conn.close()

extend_with_gin_index()

from typing import Any
import psycopg2
import matplotlib.pyplot as plt
import utils

def visualize_query_plans() -> None:
    conn: psycopg2.connection = psycopg2.connect(
        dbname="sales_db", user="postgres", password="password", host="localhost", port="5432"
    )
    cursor: psycopg2.cursor = conn.cursor()

    # Get cost for sales query
    cursor.execute("""
        EXPLAIN SELECT product, SUM(price * quantity) as total_sales
        FROM transactions
        WHERE date BETWEEN '2023-10-01' AND '2023-10-03'
        GROUP BY product
    """)
    sales_plan = cursor.fetchall()
    sales_cost = utils.parse_explain_cost(sales_plan)

    # Get cost for JSONB query
    cursor.execute("""
        EXPLAIN SELECT product, COUNT(*) as count
        FROM transactions
        WHERE metadata->'details'->>'subcategory' = 'laptop'
        GROUP BY product
    """)
    jsonb_plan = cursor.fetchall()
    jsonb_cost = utils.parse_explain_cost(jsonb_plan)

    # Plot costs
    plt.figure(figsize=(8, 6))
    plt.bar(["Sales Query", "JSONB Query"], [sales_cost, jsonb_cost])
    plt.title("Query Plan Cost Estimates")
    plt.ylabel("Cost (arbitrary units)")
    plt.grid(True)
    plt.tight_layout()
    plt.savefig("data/query_plan_plot.png", dpi=100)
    plt.close()
    print("Plot saved to data/query_plan_plot.png")

    cursor.close()
    conn.close()

visualize_query_plans()

from typing import Any
import psycopg2

def create_expression_index() -> None:
    conn: psycopg2.connection = psycopg2.connect(
        dbname="sales_db", user="postgres", password="password", host="localhost", port="5432"
    )
    cursor: psycopg2.cursor = conn.cursor()
    cursor.execute("CREATE INDEX IF NOT EXISTS idx_subcategory ON transactions ((metadata->'details'->>'subcategory'))")
    print("Expression index created")

    cursor.execute("""
        SELECT product, COUNT(*) as count
        FROM transactions
        WHERE metadata->'details'->>'subcategory' = 'laptop'
        GROUP BY product
    """)
    results: list[Any] = cursor.fetchall()
    print("Query Results:", results)

    cursor.execute("""
        EXPLAIN SELECT product, COUNT(*) as count
        FROM transactions
        WHERE metadata->'details'->>'subcategory' = 'laptop'
        GROUP BY product
    """)
    plan: list[Any] = cursor.fetchall()
    print("Query Plan:")
    for row in plan:
        print(row[0])

    conn.commit()
    cursor.close()
    conn.close()

create_expression_index()