**Video: Window Functions Challenge Introduction**

This short video introduces the challenge of using SQL window functions to analyze employee sales data. You will see how the `SUM()` OVER() and `RANK()` window functions can be used together to calculate each employee's total sales and their ranking among all employees based on those totals.

In this challenge, you will put your understanding of **window functions** into practice. You are provided with a database containing `employees` and their `sales`. Your goal is to write a SQL query that calculates each employee's total sales and their rank among all employees based on these totals. 

You should use the `SUM()` window function to compute the total sales for each employee, and the `RANK()` window function to assign a rank, where the employee with the highest total sales is ranked 1. Both calculations should be performed in a single query, and the results should include each employee's ID, name, total sales, and rank.

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 setUp(self):
        # Clean up salary_changes first to avoid FK constraint errors
        try:
            self.cursor.execute("DELETE FROM salary_changes;")
        except Exception:
            pass
        self.cursor.execute("DELETE FROM sales;")
        self.cursor.execute("DELETE FROM employees;")
        self.connection.commit()
        self.cursor.execute("""
            INSERT INTO employees (employee_id, name, department, salary) VALUES
            (1, 'Alice', 'Sales', 50000.00),
            (2, 'Bob', 'Sales', 48000.00),
            (3, 'Charlie', 'IT', 51000.00),
            (4, 'Diana', 'Sales', 47000.00);
        """)
        self.cursor.execute("""
            INSERT INTO sales (sale_id, employee_id, amount, sale_date) VALUES
            (1, 1, 1000.00, '2024-05-01'),
            (2, 1, 2000.00, '2024-05-02'),
            (3, 2, 1500.00, '2024-05-01'),
            (4, 2, 500.00, '2024-05-03'),
            (5, 3, 2500.00, '2024-05-01'),
            (6, 4, 1000.00, '2024-05-04'),
            (7, 4, 1000.00, '2024-05-05'),
            (8, 3, 500.00, '2024-05-02');
        """)
        self.connection.commit()

    def test_columns_present(self):
        with open("query.sql", "r") as file:
            query = file.read()
        self.cursor.execute(query)
        columns = [desc[0] for desc in self.cursor.description]
        expected_columns = {'employee_id', 'name', 'total_sales', 'sales_rank'}
        condition = expected_columns.issubset(set(columns))
        _dynamic_test(
            self,
            condition,
            "All required columns are present in the result.",
            f"Missing columns: {expected_columns - set(columns)}"
        )

    def test_total_sales_calculation(self):
        with open("query.sql", "r") as file:
            query = file.read()
        self.cursor.execute(query)
        rows = self.cursor.fetchall()
        columns = [desc[0] for desc in self.cursor.description]
        df = pd.DataFrame(rows, columns=columns)
        expected = {
            1: 3000.00,
            2: 2000.00,
            3: 3000.00,
            4: 2000.00
        }
        all_ok = True
        for emp_id, total in expected.items():
            filtered = df[df['employee_id'] == emp_id]
            if filtered.empty or not float(filtered.iloc[0]['total_sales']) == total:
                all_ok = False
                break
        _dynamic_test(
            self,
            all_ok,
            "Total sales for each employee are calculated correctly.",
            "Total sales calculation is incorrect for at least one employee."
        )

    def test_sales_rank(self):
        with open("query.sql", "r") as file:
            query = file.read()
        self.cursor.execute(query)
        rows = self.cursor.fetchall()
        columns = [desc[0] for desc in self.cursor.description]
        df = pd.DataFrame(rows, columns=columns)
        expected_ranks = {
            1: 1,
            3: 1,
            2: 3,
            4: 3
        }
        all_ok = True
        for emp_id, rank in expected_ranks.items():
            filtered = df[df['employee_id'] == emp_id]
            if filtered.empty or int(filtered.iloc[0]['sales_rank']) != rank:
                all_ok = False
                break
        _dynamic_test(
            self,
            all_ok,
            "Sales rank is assigned correctly using RANK().",
            "Sales rank is incorrect for at least one employee."
        )

    def test_order_by_rank(self):
        with open("query.sql", "r") as file:
            query = file.read()
        self.cursor.execute(query)
        rows = self.cursor.fetchall()
        columns = [desc[0] for desc in self.cursor.description]
        df = pd.DataFrame(rows, columns=columns)
        ranks = df['sales_rank'].astype(int).tolist()
        ordered = ranks == sorted(ranks)
        _dynamic_test(
            self,
            ordered,
            "Results are ordered by sales_rank ascending.",
            "Results are not ordered by sales_rank."
        )

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):
    import re
    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

Master advanced SQL concepts including ACID properties, query optimization, indexing, triggers, window functions, and data control. This course is designed for learners with a solid SQL foundation who want to deepen their understanding and practical skills in optimizing queries and leveraging advanced database features.

Explore the foundational ACID properties of transactions, their practical implications, and how to manage transaction isolation and consistency in SQL databases.

Learn about different types of indexes, their internal mechanisms, and how to use them for optimizing SQL queries. Explore data caching and general strategies for query performance.

Master advanced SQL features including triggers, window functions, types of window functions, Data Control Language, views, and an introduction to NoSQL.

Challenge: Using Window Function

解答