Let's assume we want to add information to the table about **which class each user belongs** to based on their account balance.    

The classes will be determined as follows:
- A person whose account balance is in the bottom `25%` will belong to the **forth class**;
- A person whose account balance is between `25%` and `50%` will belong to the **third class**;
- A person whose account balance is between `50%` and `75%` will belong to the **second class**;
- A person whose account balance is in the top `25%` will belong to the **first class**.

To solve this problem, we can use a window function called `NTILE()`.   
The `NTILE()` function is a window function that divides the rows in an **ordered partition** into a specified number of roughly equal groups, or "tiles".    
It assigns each row a group number, starting at `1`, based on the group it falls into.
```sql
NTILE(num_buckets) OVER (
    ORDER BY sort_column
)
```


In the example, we don't need to specify a `PARTITION BY` clause because we want to apply the `NTILE()` function to the **entire result set**, not to specific subsets or partitions of the data.    
The `ORDER BY` clause is sufficient as it defines the **order in which the rows are distributed into tiles**, ensuring that the grouping is based on the sorted order of the salaries across the entire table.    
If we wanted to calculate tiles within specific groups (e.g., departments), we would use `PARTITION BY`.



Note

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_ntile_applied_with_4_classes(self):
        """Check that NTILE is used with 4 classes and ordered by balance descending."""
        with open("query.sql", "r") as file:
            query = file.read()
        # Check for correct NTILE usage
        has_ntile = "NTILE(4)" in query.replace(" ", "").upper() or "NTILE (4)" in query.replace(" ", "").upper()
        has_order = "ORDER BY BALANCE DESC" in query.upper() or "ORDER BY BALANCE DESC" in query.replace("  ", " ").upper()
        _dynamic_test(
            self,
            has_ntile and has_order,
            "NTILE(4) and ORDER BY balance DESC are present.",
            "NTILE(4) or ORDER BY balance DESC is missing."
        )

    def test_columns_present(self):
        """Check that the result includes account_holder, balance, and class columns."""
        with open("query.sql", "r") as file:
            query = file.read()
        self.cursor.execute(query)
        columns = [desc[0] for desc in self.cursor.description]
        required_columns = {"account_holder", "balance", "class"}
        _dynamic_test(
            self,
            required_columns.issubset(set(columns)),
            "Result includes account_holder, balance, and class columns.",
            f"Result is missing one or more required columns: {required_columns - set(columns)}."
        )

    def test_class_values_between_1_and_4(self):
        """Check that class values are between 1 and 4 inclusive."""
        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)
        if "class" in df.columns:
            valid = df["class"].apply(lambda x: x in [1,2,3,4]).all()
        else:
            valid = False
        _dynamic_test(
            self,
            valid,
            "All class values are between 1 and 4.",
            "Some class values are not between 1 and 4."
        )

    def test_balance_descending_order(self):
        """Check that balances are sorted in descending order."""
        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)
        if "balance" in df.columns:
            sorted_correctly = df["balance"].tolist() == sorted(df["balance"].tolist(), reverse=True)
        else:
            sorted_correctly = False
        _dynamic_test(
            self,
            sorted_correctly,
            "Balances are sorted in descending order.",
            "Balances are not sorted in descending order."
        )

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

Explore advanced SQL concepts essential for backend development. Learn about ACID transaction properties, query optimization, indexing strategies, window functions, and triggers, and understand how these features help build efficient and reliable database systems used in large-scale backend applications.

Challenge: Using Window Function

Lösung