Summary  
This chapter covers multi-level data aggregation using GROUP BY with multiple columns to generate hierarchical summaries in SQL queries.

General domain of usage  
Business reporting

Multi-level aggregation is a powerful SQL technique that enables you to summarize data at more than one level of granularity in a single query. Instead of only grouping by a single field (such as product or category), you can use the `GROUP BY` clause with multiple columns to produce hierarchical summaries. This is especially important in business reporting, where decision makers need to view performance both at broader levels (like department or region) and at more detailed levels (like product category or individual product).

For example, a single-level aggregation might show total sales per product category. In contrast, a multi-level aggregation can show total sales per category and, within each category, per subcategory or product. This approach helps you identify trends and outliers at different levels of your business, supporting more informed decisions.

**Definition:** Multi-level aggregation refers to summarizing data at more than one hierarchical level (such as category and subcategory) within a single query. Unlike single-level aggregation, which summarizes by only one column, multi-level aggregation uses multiple columns with `GROUP BY`, allowing for more detailed and flexible business reporting.

Definition

When you use `GROUP BY` with multiple columns, SQL groups the data first by the first column, then by the second, and so on. This means you can see, for example, total sales for each category, broken down by subcategory. This is extremely useful for managers who want to understand not just overall performance, but also the structure of that performance across their product or business lines.

-- Aggregate total sales by category and subcategory
SELECT
    category,
    subcategory,
    SUM(total_sales) AS total_sales
FROM
    product_sales
GROUP BY
    category,
    subcategory
ORDER BY
    category,
    subcategory;

**Use Case:** The retail manager needs a report showing total sales by department and product category for the last quarter. This enables the manager to compare performance across departments and identify which categories are contributing most to revenue within each department.

-- Total sales by department and category for the last quarter (Q2 2024)
SELECT
    category AS department,
    subcategory AS product_category,
    SUM(total_sales) AS total_sales
FROM
    product_sales
WHERE
    sale_date BETWEEN '2024-04-01' AND '2024-06-30'
GROUP BY
    category,
    subcategory
ORDER BY
    category,
    subcategory;

Multi-level aggregations are not limited to categories and subcategories—they can be applied to regions, customer segments, or any other meaningful business dimension. This flexibility is what makes them so valuable for analytical reporting.

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_one_row_per_region(self):
        """Check if query returns one row per region."""
        with open("query.sql", "r") as file:
            query = file.read()
        self.cursor.execute("SELECT COUNT(DISTINCT region) FROM product_sales;")
        expected_regions = self.cursor.fetchone()[0]
        self.cursor.execute(query)
        result = self.cursor.fetchall()
        _dynamic_test(
            self,
            len(result) == expected_regions,
            "Query returns one row per region.",
            f"Query should return {expected_regions} rows (one per region), but returned {len(result)}."
        )

    def test_columns_present(self):
        """Check if result includes region and total_revenue columns."""
        with open("query.sql", "r") as file:
            query = file.read()
        self.cursor.execute(query)
        columns = [desc[0] for desc in self.cursor.description]
        expected = {"region", "total_revenue"}
        _dynamic_test(
            self,
            expected.issubset(set([normalize_text(c) for c in columns])),
            "Result includes region and total_revenue columns.",
            f"Result columns should include both 'region' and 'total_revenue'. Columns: {columns}"
        )

    def test_total_revenue_correct(self):
        """Check that total_revenue for each region is correct."""
        with open("query.sql", "r") as file:
            query = file.read()
        # Get correct sums
        self.cursor.execute("SELECT region, SUM(total_sales) FROM product_sales GROUP BY region ORDER BY region;")
        expected = dict(self.cursor.fetchall())
        # Get user result
        self.cursor.execute(query)
        result = dict(self.cursor.fetchall())
        # Compare each region
        mismatches = []
        for region in expected:
            if region not in result or float(result[region]) != float(expected[region]):
                mismatches.append((region, expected[region], result.get(region, None)))
        _dynamic_test(
            self,
            len(mismatches) == 0,
            "Total revenue for each region is correct.",
            f"Incorrect total_revenue for: {mismatches}"
        )

    def test_order_by_region(self):
        """Check that results are ordered alphabetically by region."""
        with open("query.sql", "r") as file:
            query = file.read()
        self.cursor.execute(query)
        result = self.cursor.fetchall()
        if not result:
            _dynamic_test(self, False, "", "Query returned no results.")
            return
        regions = [row[0] for row in result]
        sorted_regions = sorted(regions, key=lambda x: (x is None, x))
        _dynamic_test(
            self,
            regions == sorted_regions,
            "Results are ordered alphabetically by region.",
            f"Regions are not ordered alphabetically: {regions}"
        )

if __name__ == "__main__":
    unittest.main()

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.

Multi-level Aggregations for Business Reports

Solution