In many real-world ride-sharing applications, searching for riders based on their location and ride history is a frequent and performance-critical operation. As you have seen in previous chapters, using `EXPLAIN` helps you analyze how efficiently a query runs and highlights opportunities for optimization. Now you will apply your skills to a practical scenario involving the `riders` and `rides` tables, focusing on improving search performance for users who want to find riders from a specific pickup location who have completed at least a certain number of rides.

import unittest
import psycopg2
import pandas as pd

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_returns_expected_riders(self):
        """Check that the query returns all riders with at least one ride from '123 Main St'."""
        with open("query.sql", "r") as file:
            query = file.read()

        # Correct result: riders who have at least one ride from '123 Main St'
        solution_query = '''
        SELECT DISTINCT r.rider_id, r.first_name, r.last_name
        FROM riders r
        JOIN rides rd ON r.rider_id = rd.rider_id
        WHERE rd.pickup_location = '123 Main St'
        '''
        self.cursor.execute(solution_query)
        expected_rows = self.cursor.fetchall()
        expected_df = pd.DataFrame(expected_rows, columns=["rider_id", "first_name", "last_name"])

        self.cursor.execute(query)
        rows = self.cursor.fetchall()
        result_df = pd.DataFrame(rows, columns=["rider_id", "first_name", "last_name"])

        # Check all expected rows are present
        missing_rows = []
        for _, expected_row in expected_df.iterrows():
            expected_dict = expected_row.to_dict()
            mask = (result_df[list(expected_dict.keys())] == pd.Series(expected_dict)).all(axis=1)
            if not mask.any():
                missing_rows.append(expected_dict)
        condition = len(missing_rows) == 0
        if not condition:
            failure_msg = f"Missing expected riders: {missing_rows}"
        else:
            failure_msg = ""
        _dynamic_test(
            self,
            condition,
            "Query returns all expected riders with rides from '123 Main St'.",
            failure_msg
        )

    def test_no_unexpected_riders(self):
        """Check that the query does not return riders without rides from '123 Main St'."""
        with open("query.sql", "r") as file:
            query = file.read()
        self.cursor.execute(query)
        rows = self.cursor.fetchall()
        result_df = pd.DataFrame(rows, columns=["rider_id", "first_name", "last_name"])

        # Find all rider_ids who have at least one ride from '123 Main St'
        self.cursor.execute("SELECT DISTINCT rider_id FROM rides WHERE pickup_location = '123 Main St'")
        valid_ids = set([row[0] for row in self.cursor.fetchall()])
        unexpected = []
        for _, row in result_df.iterrows():
            if row["rider_id"] not in valid_ids:
                unexpected.append(int(row["rider_id"]))
        condition = len(unexpected) == 0
        if not condition:
            failure_msg = f"Returned unexpected rider_id(s): {unexpected}"
        else:
            failure_msg = ""
        _dynamic_test(
            self,
            condition,
            "Query does not return riders without rides from '123 Main St'.",
            failure_msg
        )

    def test_output_columns(self):
        """Check that output columns match rider_id, first_name, last_name."""
        with open("query.sql", "r") as file:
            query = file.read()
        self.cursor.execute(query)
        columns = [desc[0] for desc in self.cursor.description]
        expected = ["rider_id", "first_name", "last_name"]
        condition = columns == expected
        _dynamic_test(
            self,
            condition,
            "Query output columns are correct.",
            f"Output columns are {columns}, expected {expected}."
        )

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: Improving Rider Search Performance

Oplossing

Challenge: Improving Rider Search Performance

Oplossing