import unittest
import psycopg2
import pandas as pd
import re

class TestPostgresQuery(unittest.TestCase):
    @classmethod
    def setUpClass(cls):
        cls.connection = psycopg2.connect(
            host="localhost",
            user="agent",
            password="qwerty",
            dbname="testdb",
            sslmode="disable"
        )
        cls.cursor = cls.connection.cursor()

    def test_explain_present_before_index(self):
        """Check that the first EXPLAIN statement for the payment query is present in submission."""
        with open("query.sql", "r") as file:
            query = file.read()
        condition = (
            "explain select p.payment_id, p.amount, p.payment_method, p.payment_time, r.first_name, r.last_name"
            in normalize_text(query)
            and "from payments p" in normalize_text(query)
            and "join riders r on p.rider_id = r.rider_id" in normalize_text(query)
            and "where p.status = 'paid'" in normalize_text(query)
            and "order by p.payment_time desc" in normalize_text(query)
        )
        _dynamic_test(
            self,
            condition,
            "First EXPLAIN statement for the payment query is present.",
            "First EXPLAIN statement for the payment query is missing or incorrect."
        )

    def test_index_created(self):
        """Check that an index is created on payments(status, payment_time)."""
        with open("query.sql", "r") as file:
            query = normalize_text(file.read())
        has_index = (
            "create index" in query
            and "on payments" in query
            and "status" in query
            and "payment_time" in query
        )
        _dynamic_test(
            self,
            has_index,
            "Index on payments(status, payment_time) is created.",
            "Index on payments(status, payment_time) is not created."
        )

    def test_index_targets_both_columns(self):
        """Check that the index targets both status and payment_time columns, in order."""
        with open("query.sql", "r") as file:
            query = normalize_text(file.read())
        idx_line = None
        for line in query.split("\\n"):
            if "create index" in line and "on payments" in line:
                idx_line = line
                break
        has_both = idx_line is not None and ("(status, payment_time" in idx_line or "(status, payment_time desc" in idx_line)
        _dynamic_test(
            self,
            has_both,
            "Index targets both status and payment_time columns.",
            "Index does not target both status and payment_time columns."
        )

    def test_explain_present_after_index(self):
        """Check that the second EXPLAIN statement for the payment query is present after the index creation."""
        with open("query.sql", "r") as file:
            query = normalize_text(file.read())
        explain_count = query.count("explain select p.payment_id, p.amount, p.payment_method, p.payment_time, r.first_name, r.last_name")
        has_explain = explain_count >= 2
        _dynamic_test(
            self,
            has_explain,
            "Second EXPLAIN statement for the payment query is present after the index is created.",
            "Second EXPLAIN statement for the payment query is missing or not after index creation.")

def _dynamic_test(test_case, condition, success_message, failure_message):
    if condition:
        test_case._testMethodName = success_message
        test_case.assertTrue(True, success_message)
    else:
        test_case._testMethodName = failure_message
        test_case.fail(failure_message)

def normalize_text(text):
    text = text.lower()
    text = re.sub(r"\s{2,}", " ", text)
    text = re.sub(r"\s*([,:?])\s*", r"\1 ", text)
    return text.strip()

if __name__ == "__main__":
    unittest.main()


test_solution.py

Dive deep into the world of SQL query optimization by mastering the EXPLAIN statement and understanding how query planners work. This course blends engaging theory with hands-on, real-world tasks to help you analyze, interpret, and optimize SQL queries using EXPLAIN and related tools.

Learn the fundamentals of the EXPLAIN statement, how query planners work, and how to interpret basic query plans.

Delve into more complex query plans, including multi-table joins, subqueries, and the impact of indexes.

Apply your knowledge to complex, real-world scenarios involving large datasets, reporting queries, and performance troubleshooting.

Challenge: Diagnosing a Slow Payment Query

解答