In this advanced challenge, you will apply your knowledge of **SQL query plans** and the `EXPLAIN` command to analyze a real-world scenario involving a library's loan system. You have access to a database containing information about **authors**, **books**, **borrowers**, and their current and past loans. Your goal is to understand how a query that retrieves overdue loans is executed, interpret its plan, and suggest ways to make it more efficient. This exercise will test your ability to read and critique multi-table join plans, focusing on performance and optimization opportunities.

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_is_run(self):
        """Check if EXPLAIN output is generated for the provided query."""
        with open("query.sql", "r") as file:
            query = file.read()
        explain_present = "explain" in normalize_text(query)
        overdue_query_present = "from loans" in normalize_text(query) and "due_date < current_date" in normalize_text(query)
        _dynamic_test(
            self,
            explain_present and overdue_query_present,
            "EXPLAIN is used on the overdue loans query.",
            "EXPLAIN is not used or query does not match the overdue loans scenario."
        )

    def test_analysis_of_plan(self):
        """Check if the user analyzed the type of scans and joins in the plan."""
        with open("query.sql", "r") as file:
            query = file.read()
        normalized = normalize_text(query)
        scan_keywords = ["seq scan", "index scan"]
        join_keywords = ["hash join", "nested loop", "merge join"]
        scan_found = any(keyword in normalized for keyword in scan_keywords)
        join_found = any(keyword in normalized for keyword in join_keywords)
        _dynamic_test(
            self,
            scan_found and join_found,
            "Query plan analysis identifies scan and join types.",
            "Query plan analysis does not mention scan or join types."
        )

    def test_suggested_index(self):
        """Check if at least one relevant index or optimization is suggested."""
        with open("query.sql", "r") as file:
            query = file.read()
        normalized = normalize_text(query)
        # Look for CREATE INDEX or mention of index on relevant columns
        relevant_columns = ["due_date", "return_date", "book_id", "borrower_id"]
        index_suggested = "create index" in normalized and any(col in normalized for col in relevant_columns)
        _dynamic_test(
            self,
            index_suggested,
            "Relevant index or optimization is suggested.",
            "No relevant index or optimization is suggested."
        )

    def test_index_relevance(self):
        """Check that the suggested index is on a column used in WHERE or JOIN."""
        with open("query.sql", "r") as file:
            query = file.read()
        normalized = normalize_text(query)
        # The WHERE clause uses due_date, return_date; joins use book_id, borrower_id
        relevant = False
        if "create index" in normalized:
            if ("on loans(due_date" in normalized or
                "on loans(book_id" in normalized or
                "on loans(borrower_id" in normalized):
                relevant = True
        _dynamic_test(
            self,
            relevant,
            "Suggested index is relevant to filtering or joining columns.",
            "Suggested index is not relevant to filtering or joining columns."
        )

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: Analyzing a Library Loans Query

Solução

Challenge: Analyzing a Library Loans Query

Solução