Summary  
The chapter describes how to use SQL window functions (such as ROW_NUMBER() with PARTITION BY and ORDER BY) to assign rankings within each group and then filter those rankings to produce Top-N per-group results.

General domain of usage  
Retail sales reporting

When you need to produce a report showing the top N items within each group—such as the top 5 products by revenue in each category or the top 2 performing stores in every region—SQL window functions become your most powerful tool. A classic challenge in analytics is to answer questions like: "Who are the best-performing employees in every department?" or "Which products lead sales within each category?" These reports help businesses focus on their most successful people, products, or stores within each logical segment, guiding data-driven decisions.

To generate a Top-N-per-group report, you use a window function to assign a ranking within each group, then filter to keep only the top-ranked rows. The most common window functions for this are `ROW_NUMBER()`, `RANK()`, and `DENSE_RANK()`. You partition the data by the grouping column (such as `category`, `region`, or `department`) and order by the metric of interest (like revenue or sales). After assigning a row number within each group, you filter the results to include only the rows where the row number is less than or equal to N.

In Top-N-per-group queries, **partitioning** means dividing your data into groups (using `PARTITION BY`, such as by `category` or `region`). **Ordering** means sorting rows within each group based on a metric (such as `ORDER BY revenue DESC`), so the ranking window function can assign the correct order for each group.

Definition

SELECT
  category,
  product_name,
  total_revenue
FROM (
  SELECT
    category,
    product_name,
    SUM(total_sales) AS total_revenue,
    ROW_NUMBER() OVER (
      PARTITION BY category
      ORDER BY SUM(total_sales) DESC
    ) AS rn
  FROM product_sales
  GROUP BY category, product_name
) ranked
WHERE rn <= 5
ORDER BY category, total_revenue DESC;

**Use case:** The retail chain wants a report of the top 2 performing stores in each region.

SELECT
  region,
  total_sales,
  rn
FROM (
  SELECT
    region,
    SUM(total_sales) AS total_sales,
    ROW_NUMBER() OVER (
      PARTITION BY region
      ORDER BY SUM(total_sales) DESC
    ) AS rn
  FROM product_sales
  GROUP BY region
) ranked
WHERE rn <= 2
ORDER BY region, total_sales DESC;

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 _execute_student_query(self):
        with open("query.sql", "r") as f:
            query = f.read()
        self.cursor.execute(query)
        rows = self.cursor.fetchall()
        columns = [desc[0].lower() for desc in self.cursor.description]
        return pd.DataFrame(rows, columns=columns), query

    def test_top_3_employees_per_store(self):
        """Check that no more than 3 employees are returned per store."""
        df, _ = self._execute_student_query()
        too_many = []
        for store in df["department"].unique():
            count = (df["department"] == store).sum()
            if count > 3:
                too_many.append((store, count))
        _dynamic_test(
            self,
            len(too_many) == 0,
            "No more than 3 employees per store.",
            f"Found more than 3 employees in stores: {too_many}"
        )

    def test_sales_aggregation_and_ranking(self):
        """Check that total_sales are aggregated and ranked correctly per store."""
        df, _ = self._execute_student_query()

        self.cursor.execute("""
            SELECT e.store_id, e.first_name, e.last_name, SUM(sa.total_amount) AS total_sales
            FROM retail_employees e
            JOIN sales sa ON sa.employee_id = e.employee_id
            GROUP BY e.store_id, e.first_name, e.last_name
        """)
        expected_df = pd.DataFrame(
            self.cursor.fetchall(),
            columns=["store_id", "first_name", "last_name", "total_sales"]
        )

        self.cursor.execute("SELECT store_id, name FROM stores;")
        store_map = dict(self.cursor.fetchall())
        expected_df["department"] = expected_df["store_id"].map(store_map)
        expected_df = expected_df.sort_values(["department", "total_sales"], ascending=[True, False])
        expected_df["rn"] = expected_df.groupby("department").cumcount() + 1
        expected_df = expected_df[expected_df["rn"] <= 3].reset_index(drop=True)

        merged = pd.merge(
            df, expected_df,
            how="outer",
            on=["department", "first_name", "last_name", "rn"],
            indicator=True
        )
        missing = merged[merged["_merge"] == "right_only"]
        extra = merged[merged["_merge"] == "left_only"]
        condition = missing.empty and extra.empty
        msg = ""
        if not missing.empty:
            msg += f"Missing rows: {missing[['department','first_name','last_name','rn']].to_dict('records')} "
        if not extra.empty:
            msg += f"Extra rows: {extra[['department','first_name','last_name','rn']].to_dict('records')}"
        _dynamic_test(
            self,
            condition,
            "Sales aggregation and row numbers are correct for each top 3 per store.",
            msg
        )

    def test_employees_belong_to_correct_store(self):
        """Check that each employee is assigned to the correct store."""
        df, _ = self._execute_student_query()
        for _, row in df.iterrows():
            self.cursor.execute(
                "SELECT store_id FROM retail_employees WHERE first_name = %s AND last_name = %s",
                (row["first_name"], row["last_name"])
            )
            result = self.cursor.fetchone()
            if result is None:
                _dynamic_test(
                    self, False,
                    "",
                    f"Employee {row['first_name']} {row['last_name']} not found in retail_employees."
                )
            else:
                self.cursor.execute("SELECT name FROM stores WHERE store_id = %s", (result[0],))
                store_name = self.cursor.fetchone()[0]
                _dynamic_test(
                    self,
                    normalize_text(store_name) == normalize_text(row["department"]),
                    f"Employee {row['first_name']} {row['last_name']} assigned to correct store.",
                    f"Employee {row['first_name']} {row['last_name']} store mismatch: got '{row['department']}', expected '{store_name}'."
                )


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

Master advanced SQL techniques for building analytical reports. Learn to solve real-world business problems using multi-level aggregations, window functions, cohort and funnel analysis, KPI calculations, and more. Each chapter features theory, a realistic business use case, and a hands-on coding task.

Explore advanced SQL concepts and apply them to real-world analytical reporting scenarios across multiple business domains.

Top-N per Group Reports

Solução