Summary  
This chapter demonstrates how to pivot tabular data into cross-tabulated summaries by using CASE WHEN expressions combined with aggregation functions to conditionally generate columns.

General domain of usage  
Sales reporting

Pivot-style reporting is a powerful technique in SQL that allows you to transform row-based data into a summary table where categories become columns. This approach is particularly useful for producing cross-tabulations—tables that display the distribution of values across two or more variables. In SQL, you can achieve this by combining aggregation functions such as `SUM` or `COUNT` with `CASE WHEN` expressions. By selectively aggregating values based on conditions, you create new columns for each category or time period, making it easy to compare data side by side.

A cross-tabulation (or crosstab) is a summary table that displays the frequency or amount of data for combinations of two or more categorical variables. In business, cross-tabs are widely used to analyze sales by product and region, employee counts by department and job title, or customer counts by segment and month. They help decision-makers spot trends, outliers, and patterns that are not immediately obvious in unpivoted, row-based tables.

Definition

SELECT
  product_name,
  SUM(CASE WHEN DATE_PART('quarter', sale_date) = 1 THEN total_sales ELSE 0 END) AS q1_sales,
  SUM(CASE WHEN DATE_PART('quarter', sale_date) = 2 THEN total_sales ELSE 0 END) AS q2_sales,
  SUM(CASE WHEN DATE_PART('quarter', sale_date) = 3 THEN total_sales ELSE 0 END) AS q3_sales,
  SUM(CASE WHEN DATE_PART('quarter', sale_date) = 4 THEN total_sales ELSE 0 END) AS q4_sales
FROM product_sales
WHERE sale_date >= '2024-01-01' AND sale_date < '2025-01-01'
GROUP BY product_name
ORDER BY product_name;

Suppose the finance team requests a table that shows monthly sales for each product, with months as columns. You can use the same `CASE WHEN` logic to pivot the data by month, allowing for quick visual comparisons across products and time periods.

SELECT
  product_name,
  SUM(CASE WHEN DATE_PART('month', sale_date) = 1 THEN total_sales ELSE 0 END) AS jan_sales,
  SUM(CASE WHEN DATE_PART('month', sale_date) = 2 THEN total_sales ELSE 0 END) AS feb_sales,
  SUM(CASE WHEN DATE_PART('month', sale_date) = 3 THEN total_sales ELSE 0 END) AS mar_sales,
  SUM(CASE WHEN DATE_PART('month', sale_date) = 4 THEN total_sales ELSE 0 END) AS apr_sales,
  SUM(CASE WHEN DATE_PART('month', sale_date) = 5 THEN total_sales ELSE 0 END) AS may_sales,
  SUM(CASE WHEN DATE_PART('month', sale_date) = 6 THEN total_sales ELSE 0 END) AS jun_sales
FROM product_sales
WHERE sale_date >= '2024-01-01' AND sale_date < '2024-07-01'
GROUP BY product_name
ORDER BY product_name;

This approach can be adapted to a variety of reporting scenarios, enabling you to create highly flexible analytical summaries.

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_one_row_per_department(self):
        with open("query.sql", "r") as file:
            query = file.read()
        # Get department names
        self.cursor.execute("SELECT COUNT(*) FROM departments;")
        expected_count = self.cursor.fetchone()[0]
        self.cursor.execute(query)
        rows = self.cursor.fetchall()
        condition = len(rows) == expected_count
        _dynamic_test(
            self,
            condition,
            "Query returns one row per department.",
            f"Expected {expected_count} rows (one per department) but got {len(rows)}."
        )

    def test_manager_count(self):
        with open("query.sql", "r") as file:
            query = file.read()
        # Get correct counts
        self.cursor.execute('''
            SELECT d.name, COUNT(*) as count
            FROM departments d
            LEFT JOIN hr_employees he ON d.department_id = he.department_id
            WHERE he.position = 'Manager'
            GROUP BY d.name
        ''')
        solution = dict(self.cursor.fetchall())
        self.cursor.execute(query)
        rows = self.cursor.fetchall()
        columns = [desc[0] for desc in self.cursor.description]
        idx = columns.index('department')
        count_idx = columns.index('manager_count')
        missing = []
        for row in rows:
            dept = row[idx]
            actual = row[count_idx]
            expected = solution.get(dept, 0)
            if actual != expected:
                missing.append(f"Department: {dept}, Expected: {expected}, Got: {actual}")
        condition = len(missing) == 0
        _dynamic_test(
            self,
            condition,
            "manager_count is correct for all departments.",
            "Incorrect manager_count: " + ", ".join(missing) if missing else ""
        )

    def test_sales_rep_count(self):
        with open("query.sql", "r") as file:
            query = file.read()
        self.cursor.execute('''
            SELECT d.name, COUNT(*) as count
            FROM departments d
            LEFT JOIN hr_employees he ON d.department_id = he.department_id
            WHERE he.position = 'Sales Representative'
            GROUP BY d.name
        ''')
        solution = dict(self.cursor.fetchall())
        self.cursor.execute(query)
        rows = self.cursor.fetchall()
        columns = [desc[0] for desc in self.cursor.description]
        idx = columns.index('department')
        count_idx = columns.index('sales_rep_count')
        missing = []
        for row in rows:
            dept = row[idx]
            actual = row[count_idx]
            expected = solution.get(dept, 0)
            if actual != expected:
                missing.append(f"Department: {dept}, Expected: {expected}, Got: {actual}")
        condition = len(missing) == 0
        _dynamic_test(
            self,
            condition,
            "sales_rep_count is correct for all departments.",
            "Incorrect sales_rep_count: " + ", ".join(missing) if missing else ""
        )

    def test_software_eng_count(self):
        with open("query.sql", "r") as file:
            query = file.read()
        self.cursor.execute('''
            SELECT d.name, COUNT(*) as count
            FROM departments d
            LEFT JOIN hr_employees he ON d.department_id = he.department_id
            WHERE he.position = 'Software Engineer'
            GROUP BY d.name
        ''')
        solution = dict(self.cursor.fetchall())
        self.cursor.execute(query)
        rows = self.cursor.fetchall()
        columns = [desc[0] for desc in self.cursor.description]
        idx = columns.index('department')
        count_idx = columns.index('software_eng_count')
        missing = []
        for row in rows:
            dept = row[idx]
            actual = row[count_idx]
            expected = solution.get(dept, 0)
            if actual != expected:
                missing.append(f"Department: {dept}, Expected: {expected}, Got: {actual}")
        condition = len(missing) == 0
        _dynamic_test(
            self,
            condition,
            "software_eng_count is correct for all departments.",
            "Incorrect software_eng_count: " + ", ".join(missing) if missing else ""
        )

    def test_hr_specialist_count(self):
        with open("query.sql", "r") as file:
            query = file.read()
        self.cursor.execute('''
            SELECT d.name, COUNT(*) as count
            FROM departments d
            LEFT JOIN hr_employees he ON d.department_id = he.department_id
            WHERE he.position = 'HR Specialist'
            GROUP BY d.name
        ''')
        solution = dict(self.cursor.fetchall())
        self.cursor.execute(query)
        rows = self.cursor.fetchall()
        columns = [desc[0] for desc in self.cursor.description]
        idx = columns.index('department')
        count_idx = columns.index('hr_specialist_count')
        missing = []
        for row in rows:
            dept = row[idx]
            actual = row[count_idx]
            expected = solution.get(dept, 0)
            if actual != expected:
                missing.append(f"Department: {dept}, Expected: {expected}, Got: {actual}")
        condition = len(missing) == 0
        _dynamic_test(
            self,
            condition,
            "hr_specialist_count is correct for all departments.",
            "Incorrect hr_specialist_count: " + ", ".join(missing) if missing else ""
        )

    def test_zero_counts_for_missing_positions(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]
        # Find all possible positions
        expected_columns = ['manager_count', 'sales_rep_count', 'software_eng_count', 'hr_specialist_count']
        for row in rows:
            for col in expected_columns:
                idx = columns.index(col)
                value = row[idx]
                condition = value is not None and value >= 0
                _dynamic_test(
                    self,
                    condition,
                    f"Zero or positive count for {col} in all departments.",
                    f"Negative or NULL count for {col} in a department. Value: {value}"
                )

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.

Pivot-Style Reports with CASE WHEN

Oplossing