As you continue to develop your skills in query plan analysis, you will encounter queries that use subqueries for aggregation and filtering. These can sometimes lead to performance bottlenecks, especially on larger datasets. Understanding how to analyze and optimize such queries is essential for advanced SQL performance tuning.

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_same_result_as_original(self):
        """Check that the optimized query returns the same authors as the original query."""
        # Original query
        original_query = '''
        SELECT a.author_id, a.first_name, a.last_name
        FROM authors a
        WHERE a.author_id IN (
            SELECT b.author_id
            FROM books b
            JOIN loans l ON b.book_id = l.book_id
            GROUP BY b.author_id
            HAVING COUNT(l.loan_id) = (
                SELECT MAX(book_loan_count) FROM (
                    SELECT b2.author_id, COUNT(l2.loan_id) AS book_loan_count
                    FROM books b2
                    JOIN loans l2 ON b2.book_id = l2.book_id
                    GROUP BY b2.author_id
                ) AS author_loan_counts
            )
        );
        '''
        # Optimized query
        optimized_query = '''
        WITH author_loan_counts AS (
            SELECT b.author_id, COUNT(l.loan_id) AS book_loan_count
            FROM books b
            JOIN loans l ON b.book_id = l.book_id
            GROUP BY b.author_id
        ),
        max_loan_count AS (
            SELECT MAX(book_loan_count) AS max_count FROM author_loan_counts
        )
        SELECT a.author_id, a.first_name, a.last_name
        FROM authors a
        JOIN author_loan_counts alc ON a.author_id = alc.author_id
        JOIN max_loan_count mlc ON alc.book_loan_count = mlc.max_count;
        '''
        self.cursor.execute(original_query)
        orig_rows = self.cursor.fetchall()
        orig_columns = [desc[0] for desc in self.cursor.description]
        orig_df = pd.DataFrame(orig_rows, columns=orig_columns)

        self.cursor.execute(optimized_query)
        opt_rows = self.cursor.fetchall()
        opt_columns = [desc[0] for desc in self.cursor.description]
        opt_df = pd.DataFrame(opt_rows, columns=opt_columns)

        # Compare sets of tuples
        orig_set = set(tuple(row) for row in orig_df.to_numpy())
        opt_set = set(tuple(row) for row in opt_df.to_numpy())

        _dynamic_test(
            self,
            orig_set == opt_set,
            "Optimized query returns the same authors as the original.",
            f"Mismatch in results.\\nOriginal: {orig_set}\\nOptimized: {opt_set}"
        )

    def test_optimized_query_uses_cte(self):
        """Check that the optimized query uses a CTE for author loan counts."""
        with open("query.sql", "r") as file:
            query = file.read()
        # Look for CTE definition
        has_cte = 'with author_loan_counts as' in normalize_text(query)
        _dynamic_test(
            self,
            has_cte,
            "Optimized query uses a CTE for author loan counts.",
            "Optimized query does not use a CTE for author loan counts."
        )

    def test_optimized_query_avoids_redundant_subquery(self):
        """Check that the optimized query does not contain repeated subqueries for max loan count."""
        with open("query.sql", "r") as file:
            query = file.read()
        # Should not contain 'SELECT MAX(book_loan_count) FROM ('
        redundant = 'select max(book_loan_count) from (' in normalize_text(query)
        _dynamic_test(
            self,
            not redundant,
            "Optimized query avoids redundant subquery execution.",
            "Optimized query still contains a redundant subquery for max loan count."
        )

    def test_queries_are_valid(self):
        """Check that both queries are valid and executable."""
        # Original query
        original_query = '''
        SELECT a.author_id, a.first_name, a.last_name
        FROM authors a
        WHERE a.author_id IN (
            SELECT b.author_id
            FROM books b
            JOIN loans l ON b.book_id = l.book_id
            GROUP BY b.author_id
            HAVING COUNT(l.loan_id) = (
                SELECT MAX(book_loan_count) FROM (
                    SELECT b2.author_id, COUNT(l2.loan_id) AS book_loan_count
                    FROM books b2
                    JOIN loans l2 ON b2.book_id = l2.book_id
                    GROUP BY b2.author_id
                ) AS author_loan_counts
            )
        );
        '''
        # Optimized query
        optimized_query = '''
        WITH author_loan_counts AS (
            SELECT b.author_id, COUNT(l.loan_id) AS book_loan_count
            FROM books b
            JOIN loans l ON b.book_id = l.book_id
            GROUP BY b.author_id
        ),
        max_loan_count AS (
            SELECT MAX(book_loan_count) AS max_count FROM author_loan_counts
        )
        SELECT a.author_id, a.first_name, a.last_name
        FROM authors a
        JOIN author_loan_counts alc ON a.author_id = alc.author_id
        JOIN max_loan_count mlc ON alc.book_loan_count = mlc.max_count;
        '''
        try:
            self.cursor.execute(original_query)
            self.cursor.fetchall()
            self.cursor.execute(optimized_query)
            self.cursor.fetchall()
            valid = True
        except Exception as e:
            valid = False
        _dynamic_test(
            self,
            valid,
            "Both queries are valid and executable.",
            "One or both queries are invalid or not executable."
        )

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: Optimizing Subquery Performance

解答