Assume we want to retrieve only the account numbers that **have performed at least one transaction**.

In this task, we have a query that is **not optimized**. Your task is to rewrite it in a more optimal way.

Here is a list of the main query rewriting techniques to enhance optimization:
1. **Explicitly Specify Columns**: instead of using the asterisk (`*`) wildcard, explicitly mention column names in queries for better performance, readability, and maintainability;

2. **Minimize Subqueries**: reduce the use of subqueries to optimize query performance. Consider alternatives like joins or derived tables to avoid complexity and overhead;

3. **Avoid Repeated IN Operators**: limit the use of the `IN` operator in queries to prevent performance impact. Instead, consider using `JOIN` or `EXISTS` clauses for more efficient execution plans;

4. **Organize Joins Logically**: start SQL joins with the **main table** and then join with **related tables** to optimize query organization and database engine optimization;

5. **Use Restrictive WHERE Conditions**: improve query performance by including **restrictive conditions** in the `WHERE` clause to filter rows and enhance execution speed;

6. **Refactor Code into Stored Procedures or Functions**: encapsulate **repetitive code** segments into stored procedures or user-defined functions for code reusability, modularity, and easier maintenance. These can reduce redundancy and optimize SQL queries.

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_output_account_numbers(self):
        """Check that the output contains only correct account numbers in ascending 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]
        result_df = pd.DataFrame(rows, columns=columns)
        expected = [101, 202, 303, 404, 789]
        condition = (
            list(result_df.columns) == ["account_number"] and
            result_df["account_number"].tolist() == expected
        )
        _dynamic_test(
            self,
            condition,
            "Correct account numbers are selected and sorted ascending.",
            f"Output is not correct. Expected account numbers: {expected}, got: {result_df['account_number'].tolist()}"
        )

    def test_no_in_operator_or_subqueries(self):
        """Check that the query does not use IN operator or subqueries."""
        with open("query.sql", "r") as file:
            query = file.read()
        uses_in = " in " in query.lower() or " in(" in query.lower()
        uses_subquery = "select" in query.lower() and "(" in query
        forbidden = ("in (" in query.lower() or "in(" in query.lower())
        _dynamic_test(
            self,
            not forbidden,
            "Query does not use IN operator or subqueries.",
            "Query uses IN operator or subqueries, which is not allowed."
        )

    def test_inner_join_used(self):
        """Check that INNER JOIN is used between BankAccounts and Transactions tables."""
        with open("query.sql", "r") as file:
            query = file.read().lower().replace('\\n', ' ')
        # Check for INNER JOIN on BankAccounts and Transactions
        uses_inner_join = ("inner join transactions" in query or "inner join bankaccounts" in query)
        _dynamic_test(
            self,
            uses_inner_join,
            "INNER JOIN is used between BankAccounts and Transactions.",
            "Query does not use INNER JOIN between BankAccounts and Transactions."
        )

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: Query Optimization

Рішення