DW SQL Practice

📘 SQL Basics & Joins

Query structure & clauses

The core clause order every query follows, plus common NULL-handling functions.

-- NULL handling
SELECT IFNULL(score, 0) AS score_val FROM student;
SELECT COALESCE(score1, score2, 0) AS score_val FROM student;
-- COALESCE is more versatile — accepts multiple fallback values

💡 COALESCE > IFNULL when you need multiple fallback values, not just one.

Finding duplicates

COUNT(*) vs COUNT(column) — a subtle but important distinction.

SELECT name, COUNT(*), COUNT(student_id) -- COUNT(column) only counts non-NULL values
FROM students
GROUP BY name
HAVING COUNT(*) > 1;

💡 COUNT(*) counts all rows. COUNT(column) skips NULLs in that column.

Left Join — Anti-join pattern

Find rows in A whose key does NOT exist in B.

SELECT a.*
FROM a
LEFT JOIN b ON a.key = b.key
WHERE b.key IS NULL;

💡 This is the standard 'anti-join' pattern — LEFT JOIN + WHERE ... IS NULL.

Joins & Subqueries overview

Quick reference for the join family.

INNER JOIN -- only matching rows on both sides
LEFT JOIN -- all rows from left, matched or NULL from right
RIGHT JOIN -- all rows from right, matched or NULL from left
SELF JOIN -- a table joined to itself
CROSS JOIN -- every combination (cartesian product)

🧮 Window Functions

Window function basics

The building blocks: PARTITION BY, ORDER BY, and the functions that use them.

-- Ranking
ROW_NUMBER() -- unique rank, no ties
RANK() -- ties share rank, next rank skips
DENSE_RANK() -- ties share rank, no skip
-- Analytics
SUM(), AVG(), MIN(), MAX() OVER (...)

-- Positional
LAG() -- access the PREVIOUS row
LEAD() -- access the NEXT row
FIRST_VALUE(), LAST_VALUE()

-- Example
ROW_NUMBER() OVER (PARTITION BY user_id ORDER BY login_date)
LAG(txn_date) OVER (PARTITION BY customer_id ORDER BY txn_date)

💡 PARTITION BY = 'group by' for window functions, but rows are NOT collapsed like GROUP BY does.

💰 Aggregation, Spend & Averages

Case 1: Average DAILY spend exceeds 3

Two-step aggregation: sum per day first, then average the daily sums.

Sample data

customer_id	txn_date	txn_amount
1	2025-08-01	1.00
1	2025-08-01	2.00
1	2025-08-02	5.00
2	2025-08-01	4.00
2	2025-08-02	2.00

SELECT
  customer_id,
  AVG(daily_sum) AS average_spend
FROM (
  SELECT customer_id, txn_date, SUM(txn_amount) AS daily_sum
  FROM transaction
  GROUP BY customer_id, txn_date
) AS daily_spend
GROUP BY customer_id
HAVING AVG(daily_sum) > 3;

Result

customer_id	daily sums	average	included?
1	{3.00, 5.00}	4.00	✅ included
2	{4.00, 2.00}	3.00	❌ excluded (not > 3)

💡 Nested subquery pattern: aggregate once inside, aggregate again outside.

Case 2: Average PER TRANSACTION spend

Compare this to Case 1 — no need to group by day first.

SELECT
  customer_id,
  AVG(txn_amount) AS average_spend
FROM transaction
GROUP BY customer_id
HAVING AVG(txn_amount) > 3;

💡 Read the question carefully — 'average daily spend' and 'average per transaction' need different queries!

Average days between transactions

A classic three-step LAG pattern for measuring the gap between events.

Sample data

customer_id	txn_date	txn_amount
C1	2024-01-10	20
C1	2024-01-01	10
C1	2024-01-05	15
C2	2024-01-04	7
C2	2024-01-02	5

-- Step 1: get previous transaction date
WITH transaction_date AS (
    SELECT customer_id, txn_date,
        LAG(txn_date) OVER (PARTITION BY customer_id ORDER BY txn_date) AS prev_transaction_date
    FROM transaction
),
-- Step 2: calculate days between transactions
date_diff AS (
    SELECT customer_id,
        DATEDIFF(txn_date, prev_transaction_date) AS date_between_transactions
    FROM transaction_date
    WHERE prev_transaction_date IS NOT NULL
)
-- Step 3: average
SELECT customer_id, AVG(date_between_transactions) AS avg_days_transactions
FROM date_diff
GROUP BY customer_id;

Result

customer_id	avg_days_transactions
C1	4.5
C2	2.0

💡 LAG automatically sorts by the ORDER BY inside OVER() first, THEN takes the previous row — the input order in the raw table doesn't matter.

Top 10 spenders in last 30 days

A simple filter + aggregate + rank pattern.

SELECT user_id, SUM(amount) AS total_spend
FROM transactions
WHERE tx_time >= DATE_SUB(CURRENT_DATE(), 30)
GROUP BY user_id
ORDER BY total_spend DESC
LIMIT 10;

🔥 Consecutive Days

Consecutive 3-day logins (exact gap-of-1 method)

The classic 'row number minus date' trick to detect consecutive sequences.

Sample data

user_id	login_date
A	2025-01-01
A	2025-01-02
A	2025-01-03
A	2025-01-05

-- Step 1: assign row numbers and calculate grouping flag
WITH t2_sub_flag AS (
    SELECT user_id, login_date,
        ROW_NUMBER() OVER (PARTITION BY user_id ORDER BY login_date) AS rk,
        DATE_SUB(login_date, ROW_NUMBER() OVER (PARTITION BY user_id ORDER BY login_date)) AS flag_date
    FROM table1
)
-- Step 2: group and filter for continuous segments
, segments AS (
    SELECT user_id, flag_date, COUNT(1) AS cnt,
        MIN(login_date) AS min_login_date, MAX(login_date) AS max_login_date
    FROM t2_sub_flag
    GROUP BY user_id, flag_date
    HAVING COUNT(1) >= 3
)
-- Step 3: join back for concrete dates
SELECT s.user_id, s.flag_date, s.cnt, s.min_login_date, s.max_login_date, t.login_date
FROM segments s
JOIN table1 t ON t.user_id = s.user_id
             AND t.login_date BETWEEN s.min_login_date AND s.max_login_date
ORDER BY s.user_id, t.login_date;

Result

user_id	login_date	rk	flag_date
A	2025-01-01	1	2024-12-31
A	2025-01-02	2	2024-12-31
A	2025-01-03	3	2024-12-31
A	2025-01-05	4	2025-01-01

▶ Show execution process

💡 flag_date stays IDENTICAL for a true consecutive streak (2024-12-31 repeats 3 times) — that's the whole trick.

Variant: allow a 1-day gap

A more flexible version — tolerates small gaps instead of requiring exact consecutive dates.

Sample data

user_id	login_date
A	2025-01-01
A	2025-01-03
A	2025-01-05
A	2025-01-06

WITH login_diffs AS (
    SELECT
        user_id,
        login_date,
        DATEDIFF(
            login_date,
            LAG(login_date) OVER (
                PARTITION BY user_id
                ORDER BY login_date
            )
        ) AS diff_days
    FROM table1
),

login_flags AS (
    SELECT
        user_id,
        login_date,
        CASE
            WHEN diff_days IS NULL OR diff_days > 2 THEN 1
            ELSE 0
        END AS gap_flag
    FROM login_diffs
),

segments AS (
    SELECT
        user_id,
        login_date,
        SUM(gap_flag) OVER (
            PARTITION BY user_id
            ORDER BY login_date
            ROWS BETWEEN UNBOUNDED PRECEDING
                     AND CURRENT ROW
        ) AS segment_id
    FROM login_flags
)

SELECT
    user_id,
    segment_id,
    MIN(login_date) AS min_login_date,
    MAX(login_date) AS max_login_date,
    COUNT(*) AS login_days,
    DATEDIFF(MAX(login_date), MIN(login_date)) + 1 AS calendar_days
FROM segments
GROUP BY
    user_id,
    segment_id
HAVING COUNT(*) >= 3
ORDER BY
    user_id,
    min_login_date;

Result

user_id	segment_id	min_login_date	max_login_date	login_days	calendar_days
A	1	2025-01-01	2025-01-06	4	6

▶ Show execution process

💡 Running SUM() of a 0/1 flag is a common trick to create segment IDs — each new '1' starts a new segment.

🧩 Merge Date Ranges — Merge Intervals

Problem — Merge overlapping date ranges

Given cmp_id, start_date, and end_date, merge overlapping intervals for each campaign and return one row per merged interval.

If two ranges do not overlap, keep them as separate output rows.

This follows the same overlap rule as LeetCode 56: January 1–5 and January 6–8 do not overlap, so they remain separate.

Assume valid, non-null campaign IDs and dates, with start_date no later than end_date; validate or quarantine invalid rows first.

Memory: Sort → Previous MAX → Mark non-overlap → Running SUM → Merge.

This gaps-and-islands problem extends the preceding Consecutive Days chapter from individual dates to date ranges.

Blue = input intervals · Green = merged coverage

A1 · January 2026

Day1234567891011121314151617Row 11–5Row 24–10Row 315–17Merged10 days3 days

A2 · February 2026

Day12345678910Row 11–5Row 24–8Row 37–10Merged10 days

A1 outputs two merged intervals: January 1–10 and January 15–17.

A2 outputs one merged interval: February 1–10.

Input sample — sorted by cmp_id, start_date, end_date for readability

cmp_id	start_date	end_date
A1	2026-01-01	2026-01-05
A1	2026-01-04	2026-01-10
A1	2026-01-15	2026-01-17
A2	2026-02-01	2026-02-05
A2	2026-02-04	2026-02-08
A2	2026-02-07	2026-02-10

Sample order is only for readability. In the real case, table storage order is not guaranteed; the solution defines ORDER BY explicitly inside each window.

-- Sample is sorted only for readability.
-- Real table row order is not guaranteed.
CREATEORREPLACETEMPVIEWcampaignAS
SELECT
cmp_id,
start_date,
end_date
FROMVALUES
('A1',DATE'2026-01-01',DATE'2026-01-05'),
('A1',DATE'2026-01-04',DATE'2026-01-10'),
('A1',DATE'2026-01-15',DATE'2026-01-17'),
('A2',DATE'2026-02-01',DATE'2026-02-05'),
('A2',DATE'2026-02-04',DATE'2026-02-08'),
('A2',DATE'2026-02-07',DATE'2026-02-10')
ASdemo(cmp_id,start_date,end_date);

Expected final result

cmp_id	start_date	end_date	interval_days
A1	2026-01-01	2026-01-10	10
A1	2026-01-15	2026-01-17	3
A2	2026-02-01	2026-02-10	10

A1 must output two rows. January 15–17 does not overlap January 1–10, so it stays as a separate merged interval.

Run Steps 1–6 in order in the same Spark session, or run the complete CTE query at the end.

How to Think About It — SQL vs Python

SQL describes the grouping rules and lets the database derive the groups. Python keeps the current merged state and scans the intervals one by one.

Memory Point	SQL	Python
1. Group	`PARTITION BY cmp_id` Group rows by campaign.	`defaultdict(list)` Group intervals by cmp_id.
2. Sort	`ORDER BY start_date` Sort inside each partition.	`intervals.sort()` Sort each campaign's intervals.
3. Previous coverage	`previous_max_end` Furthest end already covered by earlier rows.	`merged[-1][1]` End of the current merged interval.
4. New group?	`CASE` If `start_date > previous_max_end` → new group.	`if start > merged[-1][1]` → append a new interval.
5. Keep state	`SUM(flag)` → `group_id` SQL stores state as derived columns.	`merged[-1]` Python keeps mutable state directly.
6. Merge	`MIN(start)` + `MAX(end)` Merge after group_id is created.	Update `merged[-1][1]` Merge while scanning.

SQL Flow — Create groups first

Memory: sort → previous_max_end → flag → group_id → MIN/MAX

Python Flow — Merge while scanning

Memory: group → sort → compare last end → append or extend

BIGGEST DIFFERENCE

SQL: create group_id first, then merge.
Python: merge while scanning.

BEST MEMORY BRIDGE

SQL previous_max_end ≈ Python merged[-1][1]
Both mean: the furthest end covered so far.

One-sentence memory: SQL = create groups, then merge. Python = merge while scanning.

Two interview versions — SQL + Python / LeetCode-style

The core idea is the same: sort intervals, decide whether the current interval connects to existing coverage, then either merge it or start a new group.

SQL Version

Data Warehouse / Spark SQL

Use the running maximum end date because SQL must handle nested ranges, not only the immediately previous row.

New group: start_date > previous_max_end

WITHbaseAS(
SELECTDISTINCT
cmp_id,
start_date,
end_date
FROMcampaign
),
orderedAS(
SELECT
cmp_id,
start_date,
end_date,

-- Step 1: previous_max_end = furthest end_date from EARLIER rows.
MAX(end_date)OVER(
PARTITIONBYcmp_id
ORDERBYstart_date,end_date
ROWSBETWEENUNBOUNDEDPRECEDINGAND1PRECEDING
)ASprevious_max_end
FROMbase
),
markedAS(
SELECT
cmp_id,
start_date,
end_date,

-- Step 2: no overlap -> start a new group.
CASE
WHENprevious_max_endISNULL
ORstart_date>previous_max_end
THEN1
ELSE0
ENDASis_new_group
FROMordered
),
groupedAS(
SELECT
cmp_id,
start_date,
end_date,

-- Step 3: cumulative flags create group_id.
-- ROWS BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW can be omitted here.
SUM(is_new_group)OVER(
PARTITIONBYcmp_id
ORDERBYstart_date,end_date
)ASgroup_id
FROMmarked
)
SELECT
cmp_id,
MIN(start_date)ASmerged_start,
MAX(end_date)ASmerged_end
FROMgrouped
GROUPBY
cmp_id,
group_id
ORDERBY
cmp_id,
merged_start;

SQL demo output:
A1 | 2026-01-01 | 2026-01-10
A1 | 2026-01-15 | 2026-01-17
A2 | 2026-02-01 | 2026-02-10

This SQL version matches LeetCode 56 semantics: merge overlapping ranges only. A non-overlapping range starts a new output row.

Python / LeetCode-style

Grouped Merge Intervals — with cmp_id

LeetCode 56 core + cmp_id grouping. First group by cmp_id, then run the standard merge logic inside each group.

Core: group by cmp_id → sort → merge → add cmp_id back

Input

records = [
    ["A1", 1, 5],
    ["A1", 4, 10],
    ["A1", 15, 17],
    ["A2", 1, 5],
    ["A2", 4, 8],
    ["A2", 7, 10],
]

Output

[
    ["A1", 1, 10],
    ["A1", 15, 17],
    ["A2", 1, 10],
]

fromcollectionsimport defaultdict
fromtypingimport List


classSolution:
    defmergeByCampaign(self, records: List[List]) -> List[List]:
        groups = defaultdict(list)

        # Step 1: group intervals by cmp_id.
        for cmp_id, start, end in records:
            groups[cmp_id].append([start, end])

        # groups now looks like:
        # {
        # "A1": [[1, 5], [4, 10], [15, 17]],
        # "A2": [[1, 5], [4, 8], [7, 10]]
        # }

        result = []

        # Step 2: merge intervals inside each cmp_id.
        for cmp_id, intervals in groups.items():
            intervals.sort(key=lambda x: x[0])
            merged = []

            for start, end in intervals:
                if not merged or start > merged[-1][1]:
                    merged.append([start, end])
                else:
                    merged[-1][1] = max(merged[-1][1], end)

            # Step 3: add cmp_id back.
            for start, end in merged:
                result.append([cmp_id, start, end])

        result.sort(key=lambda x: (x[0], x[1], x[2]))
        return result

Decision: start > merged[-1][1] → new interval; otherwise merge with max end.

Memory: SQL = previous running MAX; Python = last merged end. Both answer the same question: “Does this interval connect to coverage already built?”

📊 Sales Ranking & Top-N

Daily city sales ranking

Three window functions in one query, each with a different PARTITION/ORDER combination.

Sample data

order_id	city	order_date	sales
1	Beijing	2025-01-01	100
2	Shanghai	2025-01-01	150
3	Guangzhou	2025-01-01	120
4	Beijing	2025-01-02	300
5	Shanghai	2025-01-02	250
6	Guangzhou	2025-01-02	180
7	Beijing	2025-01-02	50

WITH aggregated AS (
  SELECT city, order_date, SUM(sales) AS daily_sales
  FROM orders
  GROUP BY city, order_date
)
SELECT
    order_date, city, daily_sales,
    ROW_NUMBER() OVER (PARTITION BY order_date ORDER BY daily_sales DESC) AS rank,
    SUM(daily_sales) OVER (PARTITION BY order_date) AS total,
    SUM(daily_sales) OVER (PARTITION BY order_date ORDER BY daily_sales DESC) AS cumulative
FROM aggregated
ORDER BY order_date, daily_sales DESC;

Result

order_date	city	daily_sales	rank	total	cumulative
2025-01-01	Shanghai	150	1	370	150
2025-01-01	Guangzhou	120	2	370	270
2025-01-01	Beijing	100	3	370	370
2025-01-02	Beijing	350	1	780	350
2025-01-02	Shanghai	250	2	780	600
2025-01-02	Guangzhou	180	3	780	780

▶ Show execution process

💡 Beijing has 2 rows on 2025-01-02 (300+50) — they get summed to 350 during the first aggregation step.

Top 3 operators per city per day

Same ranking pattern as before, but partitioned by two columns instead of one.

Sample data

order_id	city	order_date	operator_id	sales
1	Beijing	2025-01-01	101	200
2	Beijing	2025-01-01	101	300
3	Beijing	2025-01-01	102	300
4	Beijing	2025-01-01	103	250
5	Beijing	2025-01-01	104	100
6	Shanghai	2025-01-01	201	400
7	Shanghai	2025-01-01	202	300
8	Shanghai	2025-01-01	203	200
9	Shanghai	2025-01-01	204	100

WITH operator_sales AS (
    SELECT city, order_date, operator_id, SUM(sales) AS total_sales
    FROM orders
    GROUP BY city, order_date, operator_id
),
ranked AS (
    SELECT city, order_date, operator_id, total_sales,
        ROW_NUMBER() OVER (PARTITION BY city, order_date ORDER BY total_sales DESC) AS rn
    FROM operator_sales
)
SELECT city, order_date, operator_id, total_sales, rn
FROM ranked
WHERE rn <= 3
ORDER BY city, order_date, rn;

Result

city	order_date	operator_id	total_sales	rn
Beijing	2025-01-01	101	500	1
Beijing	2025-01-01	102	300	2
Beijing	2025-01-01	103	250	3
Shanghai	2025-01-01	201	400	1
Shanghai	2025-01-01	202	300	2
Shanghai	2025-01-01	203	200	3

▶ Show execution process

💡 PARTITION BY city, order_date — partitioning by TWO columns creates one ranking group per city-date combination. Operator 104 (Beijing) and 204 (Shanghai) both rank 4th and get filtered out.

Advanced SQL · Aggregation, Dates & Windows

Choose a category, inspect the input and expected output, then compare the SQL approaches.

Question Map

No.	Category	Question
1	Aggregation Basics	Just use plain GROUP BY — this is the right default
2	Aggregation Basics	When does it actually become a problem?
3	Distinct Rewrites	Split into single-dimension subqueries, then JOIN
4	Distinct Rewrites	Alternative: UNION ALL with a tag column (single scan)
5	Distinct Rewrites	When GROUP BY a, c alone is enough — and when it isn't
6	Distinct Rewrites	The leanest correct version — piggyback SUM(b) onto the dedup scan
7	Skew & Performance	Salting — for when (a,c) itself is still skewed
8	Skew & Performance	Full decision summary
9	Effective Dates	Average Selling Price · LeetCode 1251
10	Effective Dates	Product Price at a Given Date · LeetCode 1164
11	Rolling Windows	Restaurant Growth · LeetCode 1321
12	Rolling Windows	Tweets’ Rolling Averages · DataLemur
13	Combined Practice	Daily Amount, Effective Risk & Previous 3-Day Average

Aggregation Basics

Baseline queries, expected results and evidence before optimization.

Example dataset & expected results

The dataset used throughout this analysis

12 rows total. Key 'x' has 8 rows (67% of the table) — a deliberately hot key. 'y' and 'z' have 2 rows each, representing normal, evenly-sized keys.

Detail

a	b	c	d
x	100	c1	d1
x	200	c1	d2
x	150	c2	d1
x	300	c1	d3
x	250	c3	d2
x	120	c2	d1
x	180	c4	d4
x	90	c1	d2
y	50	c5	d5
y	80	c6	d5
z	30	c7	d6
z	40	c7	d7

💡 In production, 'x' wouldn't be 8 rows — it would be millions, while 'y' and 'z' stay small. The 67% concentration ratio is what matters, not the absolute row count.

The query we're optimizing, and its exact correct answer

This is the ground truth every optimized rewrite in this tab must match exactly.

SELECT a, sum(b), count(distinct c), count(distinct d)
FROM test
GROUP BY a;

Detail

a	sum(b)	count(distinct c)	count(distinct d)
x	1390	4	4
y	130	2	1
z	70	1	2

💡 Verify by hand: x has c-values {c1,c2,c3,c4} → 4 distinct. d-values {d1,d2,d3,d4} → 4 distinct. sum(b) = 100+200+150+300+250+120+180+90 = 1390.

Just use plain GROUP BY — this is the right default

Simple, readable, and correct. For this dataset (or anything small-to-medium), this is the right choice — don't reach for anything fancier without evidence of a real problem.

SELECT a, sum(b), count(distinct c), count(distinct d)
FROM test
GROUP BY a;

Detail

a	sum(b)	count(distinct c)	count(distinct d)
x	1390	4	4
y	130	2	1
z	70	1	2

💡 Premature optimization trades away readability for a performance problem that may not exist. Write this first, then measure.

When does it actually become a problem?

Only optimize once one of these is genuinely true — don't guess.

Detail

Signal	Why it matters
Table has 10M+ rows	More input and intermediate aggregation / shuffle work
c or d has high cardinality	More distinct grouping combinations and intermediate state
Data is skewed on a (like key 'x' here)	One task does disproportionately more work than others
Spark UI shows one long-tail task or an OOM	Concrete evidence, not a guess

Distinct Rewrites

Single-dimension branches, tagged expansion and combining additive metrics.

Split into single-dimension subqueries, then JOIN

Each branch isolates one aggregation dimension, but Spark may already use an efficient distinct rewrite. Compare the extra scans and joins with the native plan.

SELECT t1.a, t1.sum_b, t2.cnt_c, t3.cnt_d
FROM (SELECT a, SUM(b) AS sum_b FROM test GROUP BY a) t1
JOIN (SELECT a, COUNT(DISTINCT c) AS cnt_c FROM test GROUP BY a) t2 ON t1.a = t2.a
JOIN (SELECT a, COUNT(DISTINCT d) AS cnt_d FROM test GROUP BY a) t3 ON t1.a = t3.a;

Detail

a	sum_b	cnt_c	cnt_d
x	1390	4	4
y	130	2	1
z	70	1	2

💡 Matches the ground truth exactly. Trade-off: 3 full table scans instead of 1, plus 2 joins.

Alternative: UNION ALL with a tag column (single scan)

Tags each row with which dimension it belongs to, scans the table only once, then filters per-dimension inside the aggregate functions.

SELECT a,
  SUM(CASE WHEN tag=0 THEN b END) AS sum_b,
  COUNT(DISTINCT CASE WHEN tag=1 THEN c END) AS cnt_c,
  COUNT(DISTINCT CASE WHEN tag=2 THEN d END) AS cnt_d
FROM (
  SELECT a, b, NULL AS c, NULL AS d, 0 AS tag FROM test
  UNION ALL
SELECT a, NULL, c, NULL, 1 FROM test
  UNION ALL
SELECT a, NULL, NULL, d, 2 FROM test
) unioned
GROUP BY a;

💡 One scan instead of three, but the intermediate UNION ALL result is 3x the row count. Usually split+JOIN is easier for the optimizer to parallelize well.

When GROUP BY a, c alone is enough — and when it isn't

Grouping by the composite key (a, c) shuffles by hash(a, c), not just hash(a) — so it often self-distributes even when 'a' is a hot key. The deciding factor is c's cardinality WITHIN that hot key.

Detail

Scenario	Distinct c under hot key 'x'	Is (a,c) shuffle skewed?
This dataset	4 distinct values (c1,c2,c3,c4)	Mild — spreads across 4 shuffle partitions, not 1
Worse case	Only 1 distinct value repeated 1000s of times	Yes — every row still funnels into ONE partition

💡 In THIS dataset, GROUP BY a,c alone already spreads key 'x' across 4 partitions instead of 1 — a real improvement even before adding any salt.

The leanest correct version — piggyback SUM(b) onto the dedup scan

Instead of a separate scan just for SUM(b), compute it as a byproduct of the (a,c) aggregation — saving one full table scan compared to the 3-way split.

WITH q1 AS (
  SELECT a, c, SUM(b) AS sum_b_ac
  FROM test
  GROUP BY a, c
),
q2 AS (
  SELECT a, d
  FROM test
  GROUP BY a, d              -- no need to recompute sum(b) here
),
q1_agg AS (
  SELECT a, SUM(sum_b_ac) AS sum_b, COUNT(*) AS cnt_c
  FROM q1
  GROUP BY a
),
q2_agg AS (
  SELECT a, COUNT(*) AS cnt_d
  FROM q2
  GROUP BY a
)
SELECT q1_agg.a, q1_agg.sum_b, q1_agg.cnt_c, q2_agg.cnt_d
FROM q1_agg
JOIN q2_agg ON q1_agg.a = q2_agg.a;

Detail

a	sum_b	cnt_c	cnt_d
x	1390	4	4
y	130	2	1
z	70	1	2

💡 SUM is associative — summing partial sums grouped by (a,c) and re-summing by a gives EXACTLY the same result as summing directly by a. This cuts table scans from 3 down to 2, and the (a,c) shuffle already partially fixes skew for free.

Skew & Performance

Salting decisions, exactness and the measured escalation path.

Salting — for when (a,c) itself is still skewed

Add a deterministic hash-based salt bucket to spread the dedup work across more partitions, then sum the per-bucket counts back together.

WITH salted AS (
  SELECT a, b, c, d,
    (HASH(c) & 2147483647) % 8 AS salt_c,
    (HASH(d) & 2147483647) % 8 AS salt_d
  FROM test
),
c_dedup AS (
  SELECT DISTINCT a, salt_c, c FROM salted
),
cnt_c AS (
  SELECT a, SUM(cnt) AS cnt_c FROM (
    SELECT a, salt_c, COUNT(*) AS cnt FROM c_dedup GROUP BY a, salt_c
  ) x GROUP BY a
),
d_dedup AS (
  SELECT DISTINCT a, salt_d, d FROM salted
),
cnt_d AS (
  SELECT a, SUM(cnt) AS cnt_d FROM (
    SELECT a, salt_d, COUNT(*) AS cnt FROM d_dedup GROUP BY a, salt_d
  ) x GROUP BY a
)
SELECT a, cnt_c, cnt_d FROM cnt_c JOIN cnt_d USING (a);

Detail

a	cnt_c	cnt_d
x	4	4
y	2	1
z	1	2

⚠️ A real mistake caught while building this: using RANDOM() instead of a deterministic HASH() broke correctness on a larger test (300 distinct became 2,152 after salting). Always salt using a hash of the value itself, never a random number.

Full decision summary

The complete escalation path for this pattern, cheapest to most involved.

Detail

Level	Approach	Table scans	Use when
1	Plain GROUP BY	1	Small/medium data, no observed problem
2	Split + JOIN	3	Confirmed slowness, no skew
2b	q1/q2 combined (piggyback sum)	2	Same as above, slightly leaner
3	Salted two-stage	2 + salt overhead	Measured skew; evaluate distinct cardinality and bucket balance

💡 Always diagnose with real evidence (Spark UI task times, OOM logs) before escalating — each level trades simplicity for performance, and the trade isn't free.

Effective Dates

Inclusive price intervals and latest effective versions.

Average Selling Price · LeetCode 1251

Find each product's weighted average selling price. Join sales to the price effective on the purchase date. Round to two decimals; products without sales return zero.

Public problem reference · Sample data adapted for this handbook.

Rules & assumptions

Price intervals for a product do not overlap. Both start_date and end_date are inclusive.

UnitsSold may contain duplicate rows; the problem treats them as sales records, so do not remove them without a business rule.

Input · Prices

product_id	start_date	end_date	price
1	2026-10-01	2026-10-10	10
1	2026-10-11	2026-10-31	20
2	2026-10-01	2026-10-31	50

Input · UnitsSold

product_id	purchase_date	units
1	2026-10-05	2
1	2026-10-15	3

Expected Output

product_id	average_price
1	16.00
2	0.00

Approach + trace on the same data

1. Match dates

Product 1: October 5 matches price 10; October 15 matches price 20.

2. Compute revenue

10 × 2 + 20 × 3 = 80; total units = 5.

3. Divide and default

80 / 5 = 16.00. Product 2 has no sale, so return 0.00.

Special cases

Case	How to handle it
No sales	Keep the product using LEFT JOIN; COALESCE returns 0.
Wrong average	AVG(price) ignores the number of units sold.
Boundary	This problem includes end_date; an SCD expiry may instead be exclusive.

ID + effective interval → weighted revenue / units → default zero.

SQL · Spark SQL

-- Match both product ID and the inclusive price interval.
SELECT
    p.product_id,
    COALESCE(
        ROUND(
            SUM(p.price * s.units) / NULLIF(SUM(s.units), 0),
            2
        ),
        0
    ) AS average_price
FROM Prices p
LEFT JOIN UnitsSold s
    ON p.product_id = s.product_id
   AND s.purchase_date >= p.start_date
   AND s.purchase_date <= p.end_date
GROUP BY
    p.product_id
ORDER BY
    p.product_id;

Product Price at a Given Date · LeetCode 1164

Find each product's price on 2019-08-16. Each price change stays effective until the next change. Initial price is 10.

Public problem reference · Sample data adapted for this handbook.

Rules & assumptions

Changes on the target date are already effective. Ignore changes after the target date.

The pair (product_id, change_date) is unique. Include products whose only changes occur in the future.

Input · Products

product_id	new_price	change_date
1	20	2019-08-14
1	30	2019-08-16
1	40	2019-08-18
2	50	2019-08-15
3	80	2019-08-20

Expected Output

product_id	price
1	30
2	50
3	10

Approach + trace on the same data

1. Filter history

Keep change_date <= August 16, then take MAX(change_date) per product.

2. Look up the value

Product 1: August 16 → 30. Product 2: August 15 → 50.

3. Preserve all products

Product 3 has no effective change yet, so use initial price 10.

Special cases

Case	How to handle it
Future-only product	It must remain in the output with price 10.
Price decrease	MAX(new_price) is not the latest price.
Same date	The original key prevents two versions on the same product/date.

All IDs → latest date before cutoff → look up value → default 10.

SQL · Latest effective date, then retrieve price

WITH all_products AS (
    SELECT DISTINCT
        product_id
    FROM Products
),
latest_date AS (
    SELECT
        product_id,
        MAX(change_date) AS change_date
    FROM Products
    WHERE change_date <= '2019-08-16'
    GROUP BY
        product_id
)
SELECT
    a.product_id,
    COALESCE(p.new_price, 10) AS price
FROM all_products a
LEFT JOIN latest_date l
    ON a.product_id = l.product_id
LEFT JOIN Products p
    ON l.product_id = p.product_id
   AND l.change_date = p.change_date
ORDER BY
    a.product_id;

Rolling Windows

Daily seven-day windows versus per-user three-row windows.

Restaurant Growth · LeetCode 1321

For each complete seven-day window, return the total restaurant revenue and average daily revenue. Include the current date and previous six dates.

Public problem reference · Sample data adapted for this handbook.

Rules & assumptions

At least one customer visits every calendar day. Multiple customers may visit on the same date.

Output begins on the seventh date. The denominator is seven calendar days, not the number of customer transactions.

Input · Customer

customer_id	name	visited_on	amount
1	Alice	2026-10-01	100
2	Bob	2026-10-02	200
3	Carol	2026-10-03	300
4	David	2026-10-04	400
5	Emma	2026-10-05	500
6	Frank	2026-10-06	600
7	Grace	2026-10-07	700
8	Henry	2026-10-08	800

Expected Output

visited_on	amount	average_amount
2026-10-07	2800	400.00
2026-10-08	3500	500.00

Approach + trace on the same data

1. Daily grain

GROUP BY visited_on before using either solution.

2. First window

October 1–7: 100 + 200 + … + 700 = 2800; average = 400.

3. Slide one day

October 2–8: 2800 − 100 + 800 = 3500; average = 500.

Special cases

Case	How to handle it
Many customers per day	Aggregate first; seven raw rows do not mean seven days.
Missing dates in a variant	Create a calendar and fill zeros if required. ROWS counts rows.
Early dates	Do not emit incomplete seven-day windows in this problem.

Daily totals → current date + six preceding dates → total / 7.

Solution A · Date-range self join, no window function

WITH daily AS (
    -- Multiple customers on one date become one daily total.
    SELECT
        visited_on,
        SUM(amount) AS daily_amount
    FROM Customer
    GROUP BY
        visited_on
)
SELECT
    d.visited_on,
    SUM(h.daily_amount) AS amount,
    ROUND(SUM(h.daily_amount) / 7.0, 2) AS average_amount
FROM daily d
JOIN daily h
    ON h.visited_on >= DATE_SUB(d.visited_on, INTERVAL 6 DAY)
   AND h.visited_on <= d.visited_on
GROUP BY
    d.visited_on
HAVING COUNT(*) = 7
ORDER BY
    d.visited_on;

Solution B · Window function over daily totals

WITH daily AS (
    SELECT
        visited_on,
        SUM(amount) AS daily_amount
    FROM Customer
    GROUP BY
        visited_on
),
rolling AS (
    SELECT
        visited_on,
        SUM(daily_amount) OVER (
            ORDER BY visited_on
            ROWS BETWEEN 6 PRECEDING AND CURRENT ROW
        ) AS amount,
        COUNT(*) OVER (
            ORDER BY visited_on
            ROWS BETWEEN 6 PRECEDING AND CURRENT ROW
        ) AS days_in_window
    FROM daily
)
SELECT
    visited_on,
    amount,
    ROUND(amount / 7.0, 2) AS average_amount
FROM rolling
WHERE days_in_window = 7
ORDER BY
    visited_on;

Tweets’ Rolling Averages · DataLemur

Calculate a three-day rolling average of tweet counts for each user, rounded to two decimals.

Public problem reference · Sample data adapted for this handbook.

Rules & assumptions

This practice sample has one row per user/date and consecutive dates for each user.

Include the current day and preceding two days. At the beginning, average only the available rows.

Input · tweets

user_id	tweet_date	tweet_count
111	2026-10-01	2
111	2026-10-02	1
111	2026-10-03	3
111	2026-10-04	4
222	2026-10-01	6
222	2026-10-02	2

Expected Output

user_id	tweet_date	rolling_avg_3d
111	2026-10-01	2.00
111	2026-10-02	1.50
111	2026-10-03	2.00
111	2026-10-04	2.67
222	2026-10-01	6.00
222	2026-10-02	4.00

Approach + trace on the same data

1. Partition

User 111 and user 222 are calculated independently.

2. Order

For user 111, values are 2, 1, 3, 4 in date order.

3. Limit the frame

October 4 uses [1, 3, 4]: 8 / 3 = 2.67; October 1 has left the window.

Special cases

Case	How to handle it
No PARTITION BY	Rows from different users can mix.
Missing calendar days	ROWS counts available rows. Clarify whether to fill missing days with zero.
Initial rows	This task allows partial windows, unlike Restaurant Growth.

PARTITION BY user → ORDER BY date → current row + two previous rows.

SQL · Per-user moving average

SELECT
    user_id,
    tweet_date,
    ROUND(
        AVG(tweet_count) OVER (
            PARTITION BY user_id
            ORDER BY tweet_date
            ROWS BETWEEN 2 PRECEDING AND CURRENT ROW
        ),
        2
    ) AS rolling_avg_3d
FROM tweets
ORDER BY
    user_id,
    tweet_date;

Combined Practice

Daily totals, effective risk and a previous-three-day baseline.

Daily Amount, Effective Risk & Previous 3-Day Average

Given transaction details and a historical user-risk dimension, return each user’s daily amount, the risk effective on that date, and the average daily amount over the previous three calendar days.

Focused reconstruction based on candidate recollection, not a verified original O question. SQL dialect: Spark SQL.

Rules

A user’s risk validity intervals do not overlap. Each user/date matches at most one dimension row, so the temporal JOIN does not duplicate daily metrics.

For output date D, history is [D−3 days, D). Include all user transactions regardless of historical risk. Missing dates contribute zero; divide by 3. Assume complete history and non-NULL amounts in one currency.

Match effective_date <= D < expiry_date. NULL expiry means open-ended. Keep unmatched daily records as Unknown. Output only dates with transactions.

Input · dim_user_risk

version_id	user_id	risk_level	effective_date	expiry_date
1	U1	Low	2026-10-01	2026-10-18
2	U1	High	2026-10-18	NULL
3	U2	Medium	2026-10-01	NULL
4	U3	Low	2026-10-01	2026-10-19

version_id identifies a dimension version. With non-overlapping validity intervals, this SQL is identical with or without the version_id column; date conditions already select the unique valid version.

Input · transactions

transaction_id	user_id	transaction_date	amount
T01	U1	2026-10-17	100
T02	U1	2026-10-17	300
T03	U1	2026-10-18	200
T04	U1	2026-10-20	500
T05	U1	2026-10-20	100
T06	U2	2026-10-17	60
T07	U2	2026-10-19	120
T08	U2	2026-10-20	90
T09	U3	2026-10-20	70

Expected Output

user_id	transaction_date	risk_level	daily_amount	previous_3_days_avg_amount
U1	2026-10-17	Low	400	0.0
U1	2026-10-18	High	200	133.33
U1	2026-10-20	High	600	200.0
U2	2026-10-17	Medium	60	0.0
U2	2026-10-19	Medium	120	20.0
U2	2026-10-20	Medium	90	60.0
U3	2026-10-20	Unknown	70	0.0

Why the Risk JOIN Does Not Duplicate Rows

User/date	Low: [Oct 1, Oct 18)	High: [Oct 18, open)	Result
U1 / October 17	Matches	Does not match	One row: Low
U1 / October 18	Does not match: expiry excluded	Matches: effective date included	One row: High
U1 / October 20	Does not match	Matches	One row: High

The old version ends exactly where the new version begins, but the old expiry boundary is exclusive. Therefore October 18 does not match both versions. No ROW_NUMBER, ranked CTE or rn = 1 filter is needed.

Approach + Trace

1. Daily totals

U1: October 17 = 400, October 18 = 200, October 20 = 600.

2. Previous three dates

For U1 on October 20: (400 + 200 + 0) / 3 = 200.00. The current daily amount is 600.

3. Unique effective risk

October 17 → Low; October 18 and 20 → High. U3 on October 20 → Unknown because its risk record has expired.

Complete SQL · Same With or Without version_id

WITH daily AS (
    SELECT
        user_id,
        transaction_date,
        SUM(amount) AS daily_amount
    FROM transactions
    GROUP BY
        user_id,
        transaction_date
),
metrics AS (
    SELECT
        d.user_id,
        d.transaction_date,
        d.daily_amount,
        ROUND(
            COALESCE(SUM(h.daily_amount), 0) / 3.0,
            2
        ) AS previous_3_days_avg_amount
    FROM daily d
    LEFT JOIN daily h
        ON d.user_id = h.user_id
       AND h.transaction_date >= DATE_SUB(
            d.transaction_date,
            3
       )
       AND h.transaction_date < d.transaction_date
    GROUP BY
        d.user_id,
        d.transaction_date,
        d.daily_amount
)
SELECT
    m.user_id,
    m.transaction_date,
    COALESCE(r.risk_level, 'Unknown') AS risk_level,
    m.daily_amount,
    m.previous_3_days_avg_amount
FROM metrics m
LEFT JOIN dim_user_risk r
    ON m.user_id = r.user_id
   AND m.transaction_date >= r.effective_date
   AND (
        m.transaction_date < r.expiry_date
        OR r.expiry_date IS NULL
   )
ORDER BY
    m.user_id,
    m.transaction_date;

Recall: daily totals → self join previous three dates → direct temporal LEFT JOIN. The historical self join intentionally finds multiple dates and aggregates them; the non-overlapping risk JOIN attaches at most one risk row.

TAB 2 · SQL ↔ PYSPARK

SQL ↔ PySpark Coding Review

SQL review with editable SQL on the left and the complete corresponding PySpark solution on the right. Advanced coding patterns and quick SQL syntax notes are included below.

🔊 Click any English problem sentence, heading, or sample table cell to listen. Click the same item again, or press ■ Stop, to stop.

19 drills8 categoriesEdit + Save answersMatched SQL ↔ PySpark heights

SQL Query Execution Order

The practical SQL flow is FROM → JOIN / ON → WHERE → GROUP BY → AGG → HAVING → SELECT → ORDER BY. JOIN is optional; if there is only one table, skip that step. For outer joins, remember that the ON condition belongs to the join step, while WHERE filters the joined result.

SQL logical flow

SQL: FROM → JOIN / ON → WHERE → GROUP BY → AGG → HAVING → SELECT → ORDER BY

PySpark DataFrame flow

PYSPARK: spark.table → join → filter → groupBy → agg → filter → select → orderBy

COMMON IMPORTS

from pyspark.sql import SparkSession
from pyspark.sql import functions as F
from pyspark.sql.window import Window
from pyspark.sql.types import StringType
from pyspark import StorageLevel
spark = SparkSession.builder \
    .appName("pyspark-study-notes").getOrCreate()

HOW TO THINK

1Decide the output grain.
2Write the SQL idea first.
3Translate clause-by-clause.
4Check NULLs, duplicates, and tie-breakers.

MENTAL TRANSLATION

SQL WHEREdf.filter(...)

CASE WHENF.when(...).otherwise(...)

GROUP BYdf.groupBy(...).agg(...)

OVER (...)Window.partitionBy(...).orderBy(...)

LEFT ANTIjoin(..., how="left_anti")

EXPLODEF.explode(...)

⚡

MEMORY Think SQL first → decide output grain → translate each clause into DataFrame API. | ALIGNMENT RULE: Sample Output columns = SQL aliases = PySpark aliases.

1 · Fundamentals & Joins

4 questions

Q01 Query structure & NULL handlingTab 1 · Query structure & clauses NULL → fallback value

Problem

Practice the basic query structure and choose IFNULL / COALESCE when a value may be NULL.

Sample Input

student_id	score	score1	score2
1	NULL	NULL	85
2	90	90	80

Sample Output

student_id	score_val
1	85
2	90

SQL · LEFT

SQL AnswerEditable · Save keeps your SQL in this browser

PYSPARK · RIGHT

from pyspark.sql import functions as F
# COALESCE: first non-null value wins.
result = student.select(
    "student_id",
    F.coalesce(
        F.col("score1"),
        F.col("score2"),
        F.lit(0),
    ).alias("score_val"),
)

Q02 Finding duplicatesTab 1 GROUP BY → HAVING COUNT(*) > 1

Problem

COUNT(*) vs COUNT(column) — a subtle but important distinction.

Sample Input

student_id	name
1	Alice
2	Alice
3	Bob

Sample Output

name	row_count	student_id_count
Alice	2	2

SQL · LEFT

SQL AnswerEditable · Save keeps your SQL in this browser

PYSPARK · RIGHT

from pyspark.sql import functions as F

result = (
    students
    .groupBy("name")
    .agg(
        F.count("*").alias("row_count"),
        F.count("student_id").alias("student_id_count"),
    )
    .filter(F.col("row_count") > 1)
    .select(
        "name",
        "row_count",
        "student_id_count",
    )
)

result.show(truncate=False)

Q03 Left Join — Anti-join patternTab 1 LEFT JOIN → right key IS NULL

Problem

Given two input tables A(key, value) and B(key), return the rows from A whose key does NOT exist in B.

Sample Input

Table A

key	value
1	A1
2	A2
3	A3
4	A4

Table B

key
2
3

Sample Output

key	value
1	A1
4	A4

Keys 1 and 4 are present in A but have no matching row in B.

SQL · LEFT

SQL AnswerEditable · Save keeps your SQL in this browser

PYSPARK · RIGHT

from pyspark.sql import functions as F
# Exact LEFT JOIN + IS NULL pattern.
result = (
    a.alias("a")
    .join(
        b.alias("b"),
        F.col("a.key") == F.col("b.key"),
        "left",
    )
    .filter(F.col("b.key").isNull())
    .select(
        F.col("a.key").alias("key"),
        F.col("a.value").alias("value"),
    )
)
# Shorter equivalent in Spark:
# result = a.join(b.select("key").distinct(), on="key", how="left_anti")
result.show(truncate=False)

Q04 Joins & Subqueries overviewTab 1 INNER match · LEFT keep left · RIGHT keep right · CROSS all pairs

Problem

Choose the correct join family for the requested row set.

Sample Input

Table A	Table B
1, A1	2, B2
2, A2	3, B3

Sample Output

Join	Result idea
INNER	key 2 only
LEFT	keys 1, 2
RIGHT	keys 2, 3
CROSS	4 combinations

SQL · LEFT

SQL AnswerEditable · Save keeps your SQL in this browser

PYSPARK · RIGHT

from pyspark.sql import functions as F

# INNER JOIN: matching keys only.
inner_df = (
    a.alias("a")
    .join(b.alias("b"), F.col("a.key") == F.col("b.key"), "inner")
    .select(
        F.col("a.key").alias("key"),
        F.col("a.value").alias("a_value"),
        F.col("b.value").alias("b_value"),
    )
)

# LEFT JOIN: keep all rows from A.
left_df = (
    a.alias("a")
    .join(b.alias("b"), F.col("a.key") == F.col("b.key"), "left")
    .select(
        F.col("a.key").alias("key"),
        F.col("a.value").alias("a_value"),
        F.col("b.value").alias("b_value"),
    )
)

# RIGHT JOIN: keep all rows from B.
right_df = (
    a.alias("a")
    .join(b.alias("b"), F.col("a.key") == F.col("b.key"), "right")
    .select(
        F.col("b.key").alias("key"),
        F.col("a.value").alias("a_value"),
        F.col("b.value").alias("b_value"),
    )
)

# CROSS JOIN: every combination.
cross_df = a.alias("a").crossJoin(b.alias("b")).select(
    F.col("a.key").alias("a_key"),
    F.col("b.key").alias("b_key"),
)

# SELF JOIN: a DataFrame joined to itself.
self_df = (
    a.alias("x")
    .join(a.alias("y"), F.col("x.key") == F.col("y.key"), "inner")
    .select(
        F.col("x.key").alias("x_key"),
        F.col("y.key").alias("y_key"),
    )
)

inner_df.show(truncate=False)

2 · Aggregation, Spend & Averages

6 questions

Q05 Average DAILY spend exceeds 3Tab 1 SUM per day → AVG daily_sum → HAVING

Problem

Two-step aggregation: sum per day first, then average the daily sums.

Sample Input

customer_id	txn_date	txn_amount
1	2025-08-01	1.00
1	2025-08-01	2.00
1	2025-08-02	5.00
2	2025-08-01	4.00
2	2025-08-02	2.00

Sample Output

customer_id	average_spend
1	4.00

SQL · LEFT

SQL AnswerEditable · Save keeps your SQL in this browser

PYSPARK · RIGHT

from pyspark.sql import functions as F

# Step 1: total spend per customer per day.
daily_spend = (
    transaction
    .groupBy("customer_id", "txn_date")
    .agg(F.sum("txn_amount").alias("daily_sum"))
    .select("customer_id", "txn_date", "daily_sum") # Optional: can be omitted because groupBy + agg already returns only these columns.
)

# Step 2: average the DAILY totals, not the raw transactions.
result = (
    daily_spend
    .groupBy("customer_id")
    .agg(F.avg("daily_sum").alias("average_spend"))
    .filter(F.col("average_spend") > 3)
    .select("customer_id", "average_spend") # Optional: can be omitted because groupBy + agg already returns only these columns.
)

result.show(truncate=False)

Q06 Average PER TRANSACTION spendTab 1 AVG(txn_amount) directly

Problem

Compare this to Case 1 — no need to group by day first.

Sample Input

customer_id	txn_amount
C1	4.00
C1	6.00
C2	1.00
C2	2.00

Sample Output

customer_id	average_spend
C1	5.00

SQL · LEFT

SQL AnswerEditable · Save keeps your SQL in this browser

PYSPARK · RIGHT

from pyspark.sql import functions as F

result = (
    transaction
    .groupBy("customer_id")
    .agg(F.avg("txn_amount").alias("average_spend"))
    .filter(F.col("average_spend") > 3)
    .select("customer_id", "average_spend") # Optional: can be omitted because groupBy + agg already returns only these columns.
)

result.show(truncate=False)

Q07 Average days between transactionsTab 1 LAG → DATEDIFF → AVG

Problem

A classic three-step LAG pattern for measuring the gap between events.

Sample Input

customer_id	txn_date	txn_amount
C1	2024-01-10	20
C1	2024-01-01	10
C1	2024-01-05	15
C2	2024-01-04	7
C2	2024-01-02	5

Sample Output

customer_id	avg_days_transactions
C1	4.5
C2	2.0

SQL · LEFT

SQL AnswerEditable · Save keeps your SQL in this browser

PYSPARK · RIGHT

from pyspark.sql import functions as F
from pyspark.sql.window import Window

normalized = transaction.withColumn(
    "txn_date",
    F.to_date("txn_date"),
)

w = (
    Window
    .partitionBy("customer_id")
    .orderBy("txn_date")
)

with_prev = (
    normalized
    .withColumn(
        "prev_transaction_date",
        F.lag("txn_date").over(w),
    )
    .select(
        "customer_id",
        "txn_date",
        "prev_transaction_date",
    )
)

date_diff = (
    with_prev
    .filter(F.col("prev_transaction_date").isNotNull())
    .withColumn(
        "date_between_transactions",
        F.datediff("txn_date", "prev_transaction_date"),
    )
    .select(
        "customer_id",
        "date_between_transactions",
    )
)

result = (
    date_diff
    .groupBy("customer_id")
    .agg(
        F.avg("date_between_transactions").alias("avg_days_transactions"),
    )
    .select("customer_id", "avg_days_transactions") # Optional: can be omitted because groupBy + agg already returns only these columns.
)

result.show(truncate=False)

Q08 Top 10 spenders in last 30 daysTab 1 FILTER 30d → SUM → ORDER DESC → LIMIT 10

Problem

A simple filter + aggregate + rank pattern.

Sample Input

user_id	tx_time	amount
U1	2026-09-10	40
U1	2026-09-12	60
U2	2026-09-11	70
U3	2026-07-01	500

Sample Output

user_id	total_spend
U1	100
U2	70

SQL · LEFT

SQL AnswerEditable · Save keeps your SQL in this browser

PYSPARK · RIGHT

from pyspark.sql import functions as F

normalized = transactions.withColumn(
    "tx_date",
    F.to_date("tx_time"),
)

result = (
    normalized
    .filter(
        F.col("tx_date") >= F.date_sub(F.current_date(), 30)
    )
    .groupBy("user_id")
    .agg(F.sum("amount").alias("total_spend"))
    .select("user_id", "total_spend") # Optional: can be omitted because groupBy + agg already returns only these columns.
    .orderBy(F.col("total_spend").desc())
    .limit(10)
)

result.show(truncate=False)

Q09Latest Price vs Historical AverageGROUP BY + ROW_NUMBER + JOINAVG → latest row → threshold

Problem

Return products whose latest price is more than 50% above their historical average price. Output product_id, name, current_price, and average_price. Round the average to two decimals, then sort by average price descending and product ID ascending.

Sample Input · products

id	name	brand
101	Mate Pro X	H
102	IPhone 11	A

Sample Input · price_history

id	product_id	price	changed_at
201	102	1000.30	2023-09-01
202	101	1020.10	2023-09-15
203	101	3500.00	2023-09-20

Sample Output

product_id	name	current_price	average_price
101	Mate Pro X	3500.00	2260.05

How to think: This is a “group statistic + latest row per group” question. Calculate AVG(price) by product, rank price history by newest timestamp, keep rn = 1, join the results, then apply the 1.5× threshold.

READ THIS FIRST · key intermediate output

product_id	price	rn	average_price
101	3500.00	1	2260.05
101	1020.10	2	2260.05
102	1000.30	1	1000.30

SQL

SQL AnswerEditable · Save keeps your SQL in this browser

Q10Dual-Role Billing — Incoming & OutgoingUNION ALL + CASE + SUMnormalize roles → aggregate

Problem

Calculate monthly call units per customer. Incoming calls cost one unit per second. Outgoing calls cost 500 units for the first 120 seconds, then 2 units per additional second. Only customers with calls should appear.

Sample Input · customer_detail

id	name	phone_number
1	Robin Chase	8052564516
2	Alex McGuire	9997156377
3	Chris Blair	9922536724
4	Sam Vasquez	8068164832

Sample Input · call_record

incoming	outgoing	duration	dialed_on
8068164832	9997156377	245	2018-05-21
8052564516	9997156377	264	2018-06-10
9997156377	8068164832	258	2018-05-07

Required Output Columns

name	phone_number	call_units
The source specifies the required columns but does not provide a complete output table for the excerpted sample rows.

How to think: One call record has two business roles. Normalize both roles into the same shape phone_number | units, combine them with UNION ALL, then join customer details and aggregate.

READ THIS FIRST · role normalization

role	phone_number	units rule
incoming	incoming_number	duration
outgoing	outgoing_number	500 or 500 + (duration - 120) × 2

SQL

SQL AnswerEditable · Save keeps your SQL in this browser

3 · Window & Consecutive Days

3 questions

Q11 Window function basicsTab 1 PARTITION BY · ORDER BY · ROW_NUMBER/LAG/LEAD

Problem

The building blocks: PARTITION BY, ORDER BY, and the functions that use them.

Sample Input

user_id	login_date
A	2026-01-01
A	2026-01-03
A	2026-01-05

Sample Output

user_id	login_date	row_number	prev_login
A	2026-01-01	1	NULL
A	2026-01-03	2	2026-01-01
A	2026-01-05	3	2026-01-03

SQL · LEFT

SQL AnswerEditable · Save keeps your SQL in this browser

PYSPARK · RIGHT

from pyspark.sql import functions as F
from pyspark.sql.window import Window
base = table1.withColumn(
    "login_date",
    F.to_date("login_date"),
)
w = (
    Window
    .partitionBy("user_id")
    .orderBy("login_date")
)
result = (
    base
    .withColumn("row_number", F.row_number().over(w))
    .withColumn("prev_login", F.lag("login_date").over(w))
    .select(
        "user_id",
        "login_date",
        "row_number",
        "prev_login",
    )
)
# Other common window functions:
# F.rank().over(w)
# F.dense_rank().over(w)
# F.lead("login_date").over(w)
result.show(truncate=False)

Q12 Consecutive 3-day loginsTab 1 date - row_number = island key

Problem

The classic 'row number minus date' trick to detect consecutive sequences.

Sample Input

user_id	login_date
A	2025-01-01
A	2025-01-02
A	2025-01-03
A	2025-01-05

Sample Output

user_id	min_login_date	max_login_date	cnt
A	2025-01-01	2025-01-03	3

SQL · LEFT

SQL AnswerEditable · Save keeps your SQL in this browser

PYSPARK · RIGHT

from pyspark.sql import functions as F
from pyspark.sql.window import Window
base = (
    table1
    .select(
        "user_id",
        F.to_date("login_date").alias("login_date"),
    )
    .distinct()
)
w = (
    Window
    .partitionBy("user_id")
    .orderBy("login_date")
)
ranked = base.withColumn(
    "rn",
    F.row_number().over(w),
)
# Consecutive dates share the same login_date - rn value.
flagged = ranked.withColumn(
    "flag_date",
    F.expr("date_sub(login_date, rn)"),
)
result = (
    flagged
    .groupBy("user_id", "flag_date")
    .agg(
        F.count("*").alias("cnt"),
        F.min("login_date").alias("min_login_date"),
        F.max("login_date").alias("max_login_date"),
    )
    .filter(F.col("cnt") >= 3)
    .select(
        "user_id",
        "min_login_date",
        "max_login_date",
        "cnt",
    )
)
result.show(truncate=False)

Q13 Allow a 1-day gapTab 1 LAG → gap flag → cumulative SUM → segment

Problem

A more flexible version — tolerates small gaps instead of requiring exact consecutive dates.

Sample Input

user_id	login_date
A	2025-01-01
A	2025-01-03
A	2025-01-05
A	2025-01-06

Sample Output

user_id	segment_id	min_login_date	max_login_date	login_days	calendar_days
A	1	2025-01-01	2025-01-06	4	6

SQL · LEFT

SQL AnswerEditable · Save keeps your SQL in this browser

WITH login_diffs AS (
    SELECT
        user_id,
        login_date,
        DATEDIFF(
            login_date,
            LAG(login_date) OVER (
                PARTITION BY user_id
                ORDER BY login_date
            )
        ) AS diff_days
    FROM table1
),

login_flags AS (
    SELECT
        user_id,
        login_date,
        CASE
            WHEN diff_days IS NULL OR diff_days > 2 THEN 1
            ELSE 0
        END AS gap_flag
    FROM login_diffs
),

segments AS (
    SELECT
        user_id,
        login_date,
        SUM(gap_flag) OVER (
            PARTITION BY user_id
            ORDER BY login_date
            ROWS BETWEEN UNBOUNDED PRECEDING
                     AND CURRENT ROW
        ) AS segment_id
    FROM login_flags
)

SELECT
    user_id,
    segment_id,
    MIN(login_date) AS min_login_date,
    MAX(login_date) AS max_login_date,
    COUNT(*) AS login_days,
    DATEDIFF(MAX(login_date), MIN(login_date)) + 1 AS calendar_days
FROM segments
GROUP BY
    user_id,
    segment_id
HAVING COUNT(*) >= 3
ORDER BY
    user_id,
    min_login_date;

PYSPARK · RIGHT

from pyspark.sql import functions as F
from pyspark.sql.window import Window
w_order = Window.partitionBy("user_id").orderBy("login_date")
w_running = w_order.rowsBetween(
Window.unboundedPreceding,
Window.currentRow,
)
login_diffs = (
table1
.withColumn("prev_date", F.lag("login_date").over(w_order))
.withColumn("diff_days", F.datediff("login_date", "prev_date"))
)
login_flags = login_diffs.withColumn(
"gap_flag",
F.when(
F.col("diff_days").isNull() | (F.col("diff_days") > 2),
F.lit(1),
).otherwise(F.lit(0)),
)
segments = login_flags.withColumn(
"segment_id",
F.sum("gap_flag").over(w_running),
)
result = (
segments
.groupBy("user_id", "segment_id")
.agg(
F.min("login_date").alias("min_login_date"),
F.max("login_date").alias("max_login_date"),
F.count("*").alias("login_days"),
)
.withColumn(
"calendar_days",
F.datediff("max_login_date", "min_login_date") + F.lit(1),
)
.filter(F.col("login_days") >= 3)
.orderBy("user_id", "min_login_date")
)
result.show(truncate=False)

4 · Ranking & Top-N

2 questions

Q14 Daily city sales rankingTab 1 aggregate first → ROW_NUMBER + window SUM

Problem

Three window functions in one query, each with a different PARTITION/ORDER combination.

Sample Input

order_id	city	order_date	sales
1	Beijing	2025-01-01	100
2	Shanghai	2025-01-01	150
3	Guangzhou	2025-01-01	120
4	Beijing	2025-01-02	300
5	Shanghai	2025-01-02	250
6	Guangzhou	2025-01-02	180
7	Beijing	2025-01-02	50

Sample Output

order_date	city	daily_sales	rank	total	cumulative
2025-01-01	Shanghai	150	1	370	150
2025-01-01	Guangzhou	120	2	370	270
2025-01-01	Beijing	100	3	370	370
2025-01-02	Beijing	350	1	780	350
2025-01-02	Shanghai	250	2	780	600
2025-01-02	Guangzhou	180	3	780	780

SQL · LEFT

SQL AnswerEditable · Save keeps your SQL in this browser

PYSPARK · RIGHT

from pyspark.sql import functions as F
from pyspark.sql.window import Window
aggregated = (
    orders
    .groupBy("city", "order_date")
    .agg(F.sum("sales").alias("daily_sales"))
)
w_rank = (
    Window
    .partitionBy("order_date")
    .orderBy(F.col("daily_sales").desc())
)
w_total = Window.partitionBy("order_date")
w_cumulative = w_rank.rowsBetween(
    Window.unboundedPreceding,
    Window.currentRow,
)
result = (
    aggregated
    .withColumn("rank", F.row_number().over(w_rank))
    .withColumn("total", F.sum("daily_sales").over(w_total))
    .withColumn("cumulative", F.sum("daily_sales").over(w_cumulative))
    .select(
        "order_date",
        "city",
        "daily_sales",
        "rank",
        "total",
        "cumulative",
    )
    .orderBy("order_date", F.col("daily_sales").desc())
)
result.show(truncate=False)

Q15 Top 3 operators per city per dayTab 1 GROUP BY → ROW_NUMBER per city/day → rn <= 3

Problem

Same ranking pattern as before, but partitioned by two columns instead of one.

Sample Input

order_id	city	order_date	operator_id	sales
1	Beijing	2025-01-01	101	200
2	Beijing	2025-01-01	101	300
3	Beijing	2025-01-01	102	300
4	Beijing	2025-01-01	103	250
5	Beijing	2025-01-01	104	100
6	Shanghai	2025-01-01	201	400
7	Shanghai	2025-01-01	202	300
8	Shanghai	2025-01-01	203	200
9	Shanghai	2025-01-01	204	100

Sample Output

city	order_date	operator_id	total_sales	rn
Beijing	2025-01-01	101	500	1
Beijing	2025-01-01	102	300	2
Beijing	2025-01-01	103	250	3
Shanghai	2025-01-01	201	400	1
Shanghai	2025-01-01	202	300	2
Shanghai	2025-01-01	203	200	3

SQL · LEFT

SQL AnswerEditable · Save keeps your SQL in this browser

PYSPARK · RIGHT

from pyspark.sql import functions as F
from pyspark.sql.window import Window
operator_sales = (
    orders
    .groupBy("city", "order_date", "operator_id")
    .agg(F.sum("sales").alias("total_sales"))
)
w = (
    Window
    .partitionBy("city", "order_date")
    .orderBy(F.col("total_sales").desc())
)
result = (
    operator_sales
    .withColumn("rn", F.row_number().over(w))
    .filter(F.col("rn") <= 3)
    .select(
        "city",
        "order_date",
        "operator_id",
        "total_sales",
        "rn",
    )
    .orderBy("city", "order_date", "rn")
)
result.show(truncate=False)

5 · Time Overlap, Interval Merge & Duration

3 questions

Fast review: expand one question at a time. Read the key intermediate-output table first, then compare the SQL and PySpark code.

Q16 Merge overlapping date rangesTime Pattern · Interval Merge SORT → previous MAX(end) → flag → running SUM → MIN/MAX

Problem

Given cmp_id, start_date, and end_date, merge overlapping intervals for each campaign and return one row per merged interval.

Sample Input

cmp_id	start_date	end_date
A1	2026-01-01	2026-01-05
A1	2026-01-04	2026-01-10
A1	2026-01-15	2026-01-17
A2	2026-02-01	2026-02-05
A2	2026-02-04	2026-02-08
A2	2026-02-07	2026-02-10

Sample Output

cmp_id	start_date	end_date	interval_days
A1	2026-01-01	2026-01-10	10
A1	2026-01-15	2026-01-17	3
A2	2026-02-01	2026-02-10	10

How to think · Q16 Merge Overlapping Date Ranges

Do not compare only with the immediately previous interval. Track the furthest end reached by all previous intervals, then start a new group only when the current start is beyond that boundary.

SORT → PREV MAX → FLAG → GROUP → MERGE

1 · SORTPartition by cmp_id and order by start_date, end_date.

2 · PREVIOUS MAX ENDMAX(end_date) over previous rows only handles nested overlap safely.

3 · FLAG + GROUPstart_date > previous_max_end means a new island; cumulative SUM creates group_id.

4 · MERGEWithin each group, MIN(start_date) and MAX(end_date) produce one merged range.

Key intermediate output · this is the table to understand

cmp_id	start_date	end_date	previous_max_end	new_group	group_id
A1	2026-01-01	2026-01-05	NULL	1	1
A1	2026-01-04	2026-01-10	2026-01-05	0	1
A1	2026-01-15	2026-01-17	2026-01-10	1	2
A2	2026-02-01	2026-02-05	NULL	1	1
A2	2026-02-04	2026-02-08	2026-02-05	0	1
A2	2026-02-07	2026-02-10	2026-02-08	0	1

Memory point: Q16 is the canonical merge-interval / islands-and-gaps pattern. The business entity can change, but the algorithm stays the same.

SQL · LEFT

SQL AnswerEditable · Save keeps your SQL in this browser

WITH base AS (
    SELECT DISTINCT
        cmp_id,
        start_date,
        end_date
    FROM campaign
),
ordered AS (
    SELECT
        cmp_id,
        start_date,
        end_date,
        MAX(end_date) OVER (
            PARTITION BY cmp_id
            ORDER BY start_date, end_date
            ROWS BETWEEN UNBOUNDED PRECEDING AND 1 PRECEDING
        ) AS previous_max_end
    FROM base
),
marked AS (
    SELECT
        cmp_id,
        start_date,
        end_date,
        CASE
            WHEN previous_max_end IS NULL
                OR start_date > previous_max_end
            THEN 1
            ELSE 0
        END AS new_group
    FROM ordered
),
grouped AS (
    SELECT
        cmp_id,
        start_date,
        end_date,
        SUM(new_group) OVER (
            PARTITION BY cmp_id
            ORDER BY start_date, end_date
        ) AS group_id
    FROM marked
),
merged AS (
    SELECT
        cmp_id,
        group_id,
        MIN(start_date) AS start_date,
        MAX(end_date) AS end_date
    FROM grouped
    GROUP BY cmp_id, group_id
)
SELECT
    cmp_id,
    start_date,
    end_date,
    DATEDIFF(end_date, start_date) + 1 AS interval_days
FROM merged
ORDER BY cmp_id, start_date;

PYSPARK · RIGHT

from pyspark.sql import functions as F
from pyspark.sql.window import Window
base = (
    campaign
    .select(
        "cmp_id",
        F.to_date("start_date").alias("start_date"),
        F.to_date("end_date").alias("end_date"),
    )
    .distinct()
)
w_order = (
    Window
    .partitionBy("cmp_id")
    .orderBy("start_date", "end_date")
)
w_prev = w_order.rowsBetween(
    Window.unboundedPreceding,
    -1,
)
ordered = base.withColumn(
    "previous_max_end",
    F.max("end_date").over(w_prev),
)
marked = ordered.withColumn(
    "new_group",
    F.when(
        F.col("previous_max_end").isNull()
        | (F.col("start_date") > F.col("previous_max_end")),
        F.lit(1),
    ).otherwise(F.lit(0)),
)
w_running = w_order.rowsBetween(
    Window.unboundedPreceding,
    Window.currentRow,
)
grouped = marked.withColumn(
    "group_id",
    F.sum("new_group").over(w_running),
)
result = (
    grouped
    .groupBy("cmp_id", "group_id")
    .agg(
        F.min("start_date").alias("start_date"),
        F.max("end_date").alias("end_date"),
    )
    .withColumn(
        "interval_days",
        F.datediff("end_date", "start_date") + F.lit(1),
    )
    .select("cmp_id", "start_date", "end_date", "interval_days")
    .orderBy("cmp_id", "start_date")
)
result.show(truncate=False)

Q17 Maximum Overlap / Minimum PlatformsTime Pattern · Peak Count EVENTS → NET → RUN → MAX

Problem

Q17 THINK · Interval is not a daily table. Turn boundaries into signed events, net equal timestamps, build a running state, then take the peak.

Sample Input

policy_id	effective_from	effective_to
P1	2026-01-01	2026-01-10
P2	2026-01-04	2026-01-12
P3	2026-01-07	2026-01-09
P4	2026-01-12	2026-01-15

Sample Output

max_active_policies
3

How to think · Q17 Maximum Overlap

The requested output is one peak number. Convert interval boundaries into signed events, collapse events at the same timestamp, calculate the running active count, then take MAX.

EVENTS → NET → RUN → MAX

1 · EVENTIZEeffective_from = +1; effective_to = -1.

2 · NET SAME TIMESUM(delta) at the same timestamp before the running sum.

3 · RUNNING STATECumulative SUM(delta) is the active policy count after each boundary.

4 · PEAKMAX(active_count) is the maximum overlap.

READ THIS FIRST · Key intermediate output · running active state

event_time	net delta	active_count
2026-01-01	+1	1
2026-01-04	+1	2
2026-01-07	+1	3
2026-01-09	-1	2
2026-01-10	-1	1
2026-01-12	0	1
2026-01-15	-1	0

At 2026-01-12, one policy ends and another starts. Under the half-open [start, end) rule, -1 + 1 nets to 0 before the running SUM.

SQL · LEFT

SQL AnswerEditable · Save keeps your SQL in this browser

PYSPARK · RIGHT

from pyspark.sql import functions as F
from pyspark.sql.window import Window
normalized = (
    policy
    .select(
        "policy_id",
        F.to_date("effective_from").alias("effective_from"),
        F.to_date("effective_to").alias("effective_to"),
    )
)
starts = normalized.select(
    F.col("effective_from").alias("event_time"),
    F.lit(1).alias("delta"),
)
ends = normalized.select(
    F.col("effective_to").alias("event_time"),
    F.lit(-1).alias("delta"),
)
raw_events = starts.unionByName(ends)
events = (
    raw_events
    .groupBy("event_time")
    .agg(F.sum("delta").alias("delta"))
)
w = (
    Window
    .orderBy("event_time")
    .rowsBetween(
        Window.unboundedPreceding,
        Window.currentRow,
    )
)
running = events.withColumn(
    "active_count",
    F.sum("delta").over(w),
)
result = running.agg(
    F.max("active_count").alias("max_active_policies")
)
result.show(truncate=False)

Q18 START/END Pairing + DurationTime Pattern · Duration DEDUPE STATE → LEAD → VALID PAIR → DURATION → SUM

Problem

Q18 THINK · Raw status logs are not durations. Remove duplicate consecutive states first, pair each real state with the next boundary, then sum only ACTIVE periods.

Sample Input

policy_id	status	status_time
P1	ACTIVE	2026-01-01
P1	ACTIVE	2026-01-02
P1	SUSPENDED	2026-01-05
P1	ACTIVE	2026-01-07
P1	LAPSED	2026-01-12
P2	ACTIVE	2026-01-03
P2	LAPSED	2026-01-10

Sample Output

policy_id	total_active_days
P1	9
P2	7

How to think · Q18 Pairing + Duration

The raw log is a sequence of states, not durations. First remove duplicate consecutive states, then pair each real state change with the next boundary and keep only ACTIVE periods.

ORDER → DEDUPE → PAIR → DURATION → SUM

1 · ORDER + LAGCompare each status with the previous status for the same policy.

2 · DEDUPEConsecutive duplicate states do not create a new business boundary.

3 · LEADThe next real status_time becomes the end_time of the current period.

4 · DURATIONKeep ACTIVE rows, subtract timestamps, then SUM per policy.

READ THIS FIRST · Key output 1 · after removing duplicate states

policy_id	status	status_time
P1	ACTIVE	2026-01-01
P1	SUSPENDED	2026-01-05
P1	ACTIVE	2026-01-07
P1	LAPSED	2026-01-12
P2	ACTIVE	2026-01-03
P2	LAPSED	2026-01-10

Key output 2 · ACTIVE periods after LEAD pairing

policy_id	start_time	end_time	days
P1	2026-01-01	2026-01-05	4
P1	2026-01-07	2026-01-12	5
P2	2026-01-03	2026-01-10	7

Why Q18 is harder than a simple DATEDIFF: the interval boundaries must be reconstructed correctly from noisy state logs first.

SQL · LEFT

SQL AnswerEditable · Save keeps your SQL in this browser

-- MEMORY: ORDER -> DEDUPE -> PAIR -> DURATION -> SUM
WITH ordered AS (
    SELECT
        policy_id,
        status,
        status_time,
        LAG(status) OVER (
            PARTITION BY policy_id
            ORDER BY status_time
        ) AS prev_status
    FROM policy_status_log
),
state_changes AS (
    SELECT
        policy_id,
        status,
        status_time
    FROM ordered
    WHERE prev_status IS NULL
        OR status <> prev_status
),
paired AS (
    SELECT
        policy_id,
        status,
        status_time AS start_time,
        LEAD(status_time) OVER (
            PARTITION BY policy_id
            ORDER BY status_time
        ) AS end_time
    FROM state_changes
),
active_periods AS (
    SELECT
        policy_id,
        start_time,
        end_time,
        DATEDIFF(end_time, start_time) AS active_days
    FROM paired
    WHERE status = 'ACTIVE'
        AND end_time IS NOT NULL
        AND end_time > start_time
)
SELECT
    policy_id,
    SUM(active_days) AS total_active_days
FROM active_periods
GROUP BY policy_id
ORDER BY policy_id;

PYSPARK · RIGHT

from pyspark.sql import functions as F
from pyspark.sql.window import Window

normalized = policy_status_log.withColumn(
    "status_time",
    F.to_timestamp("status_time"),
)

w_order = (
    Window
    .partitionBy("policy_id")
    .orderBy("status_time")
)

ordered = normalized.withColumn(
    "prev_status",
    F.lag("status").over(w_order),
)

state_changes = (
    ordered
    .filter(
        F.col("prev_status").isNull()
        | (F.col("status") != F.col("prev_status"))
    )
    .select("policy_id", "status", "status_time")
)

paired = (
    state_changes
    .withColumn(
        "end_time",
        F.lead("status_time").over(w_order),
    )
    .select(
        "policy_id",
        "status",
        F.col("status_time").alias("start_time"),
        "end_time",
    )
)

active_periods = (
    paired
    .filter(
        (F.col("status") == "ACTIVE")
        & F.col("end_time").isNotNull()
        & (F.col("end_time") > F.col("start_time"))
    )
    .withColumn(
        "active_days",
        F.datediff("end_time", "start_time"),
    )
)

result = (
    active_periods
    .groupBy("policy_id")
    .agg(F.sum("active_days").alias("total_active_days"))
    .orderBy("policy_id")
)

result.show(truncate=False)

Policy Data · Latest State & Customer Summary

1 drill · Q19

Q19Latest Policy + Customer SummaryMock practice · ROW_NUMBER + FILTER + GROUP BY + LEFT JOINLATEST → ACTIVE → SUM → LEFT JOIN

Problem

Mock insurance-data practice; this is not a confirmed Great Eastern assessment question. For each policy, select its latest record by updated_at, breaking ties by version. Using 2026-10-09 as the reference date, return every customer's active policy count and total annual premium, including customers with zero active policies.

An active policy has status ACTIVE, effective_date on or before the reference date, and expiry_date on or after it (or NULL). Date boundaries are inclusive. Treat NULL premiums as zero. Assume customer_id is unique in customers; (policy_id, updated_at, version) is unique in policy_history, with non-NULL ordering fields. The latest record defines the current state; updated_at is a version timestamp, not an as-of filter.

Understand the Problem · What Are We Counting?

We need one final row per customer, showing how many policies are currently active and their total annual premium.

What is a policy?A policy is an insurance contract. policy_id identifies the contract; customer_id identifies its owner. annual_premium is the yearly premium amount.

Why are there multiple rows?policy_history contains versions of the same contract. P1 appears twice because its status changed from ACTIVE to CANCELLED. Those rows are two versions of one policy, not two policies.

Which version should we use?Use the greatest updated_at for each policy_id. If timestamps tie, use the greatest version. For P2, choose version 2 with premium 2200; do not add its old premium 2000.

What does active mean?The latest status must be ACTIVE, the policy must have started by 2026-10-09, and it must not have expired before that date. NULL expiry_date means no expiry limit for this exercise.

Two different dates: updated_at decides which version is latest. effective_date and expiry_date decide whether that selected policy covers the reference date. This exercise uses the latest available state; it does not reconstruct a historical snapshot.

Read the sample: Alice owns P1 and P2. P1 is cancelled, so only P2 counts: 1 policy and 2200 premium. Bob owns P3 and P4. P3 expired on September 30; P4 starts on October 9 and has no expiry date, so Bob gets 1 policy and 800 premium. Carol owns no policy, but the problem asks for every customer, so she gets 0 and 0.

Why this order? Latest version → active policies → customer totals → all customers. Filtering ACTIVE too early incorrectly revives P1's old record. Joining customers at the end preserves Carol without inflating the policy count.

Sample Input · policy_history

policy_id	customer_id	version	status	effective_date	expiry_date	annual_premium	updated_at
P1	C1	1	ACTIVE	2026-01-01	2026-12-31	1000	2026-01-01
P1	C1	2	CANCELLED	2026-01-01	2026-12-31	1000	2026-10-01
P2	C1	1	ACTIVE	2026-01-01	2026-12-31	2000	2026-02-01
P2	C1	2	ACTIVE	2026-01-01	2026-12-31	2200	2026-09-01
P3	C2	1	ACTIVE	2025-01-01	2026-09-30	1500	2026-09-01
P4	C2	1	ACTIVE	2026-10-09	NULL	800	2026-10-08

customers

customer_id	customer_name
C1	Alice
C2	Bob
C3	Carol

Sample Output

customer_id	customer_name	active_policy_count	total_annual_premium
C1	Alice	1	2200
C2	Bob	1	800
C3	Carol	0	0

P1 is cancelled in its latest version. P2 contributes 2200. P3 has expired. P4 starts on the reference date and contributes 800. C3 has no policy and remains in the output.

How to think · Q19 Latest Policy

Find the latest state before deciding whether a policy is active.

LATEST → FILTER → AGGREGATE → JOIN

1 · Rank the complete historyPartition by policy_id. Sort updated_at descending, then version descending. Keep rn = 1.

2 · Apply active-date rulesRequire ACTIVE, effective_date ≤ reference date, and expiry_date ≥ reference date or NULL. Both date boundaries are inclusive.

3 · Aggregate by customerCount active policies and sum annual premiums. Treat a NULL premium as zero.

4 · Preserve every customerLEFT JOIN the summary to customers and replace missing counts and totals with zero.

Common trap: Filtering ACTIVE before ROW_NUMBER incorrectly keeps P1's older ACTIVE record. Rank all versions first. Aggregating before the LEFT JOIN also avoids accidentally counting a customer with no policy as one.

Interview English: I first select the latest record for each policy, including cancelled records. Then I filter policies that are active on the reference date, aggregate by customer, and left join the result to all customers. I test cancellation, expired policies, inclusive date boundaries, NULL expiry dates, NULL premiums, and customers with no active policies.

Follow-up: For a historical snapshot, clarify whether records updated after the reference timestamp should be excluded before ranking. At scale, inspect the window shuffle and partition skew; broadcast the customer table only if it is sufficiently small.

Key Intermediate Outputs · Follow the Same Sample

1 · ranked_policies — all six rows

Sort within each policy, newest first. rn = 1 marks the latest version. Do not filter ACTIVE yet.

policy_id	version	status	updated_at	rn
P1	2	CANCELLED	2026-10-01	1
P1	1	ACTIVE	2026-01-01	2
P2	2	ACTIVE	2026-09-01	1
P2	1	ACTIVE	2026-02-01	2
P3	1	ACTIVE	2026-09-01	1
P4	1	ACTIVE	2026-10-08	1

2 · latest_policies — four rows

Keep rn = 1. P1 stays in this intermediate result as CANCELLED; its older ACTIVE record is gone.

policy_id	customer_id	version	status	premium
P1	C1	2	CANCELLED	1000
P2	C1	2	ACTIVE	2200
P3	C2	1	ACTIVE	1500
P4	C2	1	ACTIVE	800

3 · active-date check — reference date 2026-10-09

Check status AND start date AND expiry date. All three conditions must pass.

policy_id	status ACTIVE?	start ≤ date?	end ≥ date or NULL?	keep?
P1	No	Yes	Yes	No — cancelled
P2	Yes	Yes	Yes	Yes
P3	Yes	Yes	No	No — expired Sep 30
P4	Yes	Yes — starts today	Yes — NULL	Yes

4 · active_policies — two rows

Only P2 and P4 contribute to the count and premium.

policy_id	customer_id	annual_premium
P2	C1	2200
P4	C2	800

5 · customer_summary — two rows

GROUP BY customer_id. C3 has no active policy, so C3 is absent at this stage.

customer_id	active_policy_count	total_annual_premium
C1	1	2200
C2	1	800

6 · LEFT JOIN + zero fill — final three rows

Start from all customers. C3 gets NULL summary fields after the join; COALESCE / fillna converts them to zero.

customer_id	customer_name	count after join	premium after join	final count	final premium
C1	Alice	1	2200	1	2200
C2	Bob	1	800	1	800
C3	Carol	NULL	NULL	0	0

Row-count memory: 6 history rows → 4 latest policies → 2 active policies → 2 customer summaries → 3 final customers. SQL and PySpark follow exactly the same stages.

SQL · LEFT

SQL AnswerEditable · Save keeps your SQL in this browser

-- 1. Rank ALL records before filtering status.
WITH ranked_policies AS (
    SELECT p.*,
           ROW_NUMBER() OVER (
               PARTITION BY policy_id
               ORDER BY updated_at DESC, version DESC
           ) AS rn
    FROM policy_history p
),
-- 2. Keep the latest record if the policy is active today.
active_policies AS (
    SELECT policy_id, customer_id, annual_premium
    FROM ranked_policies
    WHERE rn = 1
      AND status = 'ACTIVE'
      AND effective_date <= DATE '2026-10-09'
      AND (expiry_date >= DATE '2026-10-09'
           OR expiry_date IS NULL)
),
-- 3. Aggregate at customer level.
customer_summary AS (
    SELECT customer_id,
           COUNT(*) AS active_policy_count,
           SUM(COALESCE(annual_premium, 0)) AS total_annual_premium
    FROM active_policies
    GROUP BY customer_id
)
-- 4. Preserve customers with no active policies.
SELECT c.customer_id,
       c.customer_name,
       COALESCE(s.active_policy_count, 0) AS active_policy_count,
       COALESCE(s.total_annual_premium, 0) AS total_annual_premium
FROM customers c
LEFT JOIN customer_summary s
    ON c.customer_id = s.customer_id
ORDER BY c.customer_id;

PYSPARK · RIGHT

from pyspark.sql import functions as F
from pyspark.sql.window import Window

# Runnable sample: create the two input DataFrames first.
policy_rows = [
    ("P1", "C1", 1, "ACTIVE", "2026-01-01", "2026-12-31", 1000, "2026-01-01"),
    ("P1", "C1", 2, "CANCELLED", "2026-01-01", "2026-12-31", 1000, "2026-10-01"),
    ("P2", "C1", 1, "ACTIVE", "2026-01-01", "2026-12-31", 2000, "2026-02-01"),
    ("P2", "C1", 2, "ACTIVE", "2026-01-01", "2026-12-31", 2200, "2026-09-01"),
    ("P3", "C2", 1, "ACTIVE", "2025-01-01", "2026-09-30", 1500, "2026-09-01"),
    ("P4", "C2", 1, "ACTIVE", "2026-10-09", None, 800, "2026-10-08"),
]
policy_history = spark.createDataFrame(
    policy_rows,
    "policy_id string, customer_id string, version int, status string, "
    "effective_date string, expiry_date string, annual_premium long, updated_at string"
)
customers = spark.createDataFrame(
    [("C1", "Alice"), ("C2", "Bob"), ("C3", "Carol")],
    "customer_id string, customer_name string"
)

as_of = F.to_date(F.lit("2026-10-09"))

# 1. Normalize types before sorting and comparing dates.
policies = (
    policy_history
    .withColumn("effective_date", F.to_date("effective_date"))
    .withColumn("expiry_date", F.to_date("expiry_date"))
    .withColumn("updated_at", F.to_timestamp("updated_at"))
    .withColumn("version", F.col("version").cast("int"))
    .withColumn("annual_premium",
                F.col("annual_premium").cast("decimal(18,2)"))
)

# 2. Rank ALL statuses. P1's latest record is CANCELLED.
latest_window = Window.partitionBy("policy_id").orderBy(
    F.col("updated_at").desc(),
    F.col("version").desc()
)
ranked_policies = policies.withColumn(
    "rn", F.row_number().over(latest_window)
)
latest_policies = ranked_policies.filter(F.col("rn") == 1)

# 3. Only P2 and P4 remain active on the reference date.
active_policies = latest_policies.filter(
    (F.col("status") == "ACTIVE")
    & (F.col("effective_date") <= as_of)
    & ((F.col("expiry_date") >= as_of)
       | F.col("expiry_date").isNull())
)

# 4. One summary row per customer with active policies.
customer_summary = active_policies.groupBy("customer_id").agg(
    F.count("*").alias("active_policy_count"),
    F.sum(F.coalesce(F.col("annual_premium"), F.lit(0)))
     .alias("total_annual_premium")
)

# 5. LEFT JOIN retains C3. Replace missing totals with zero.
result = (
    customers
    .join(customer_summary, on="customer_id", how="left")
    .fillna({"active_policy_count": 0, "total_annual_premium": 0})
    .select("customer_id", "customer_name",
            "active_policy_count", "total_annual_premium")
    .orderBy("customer_id")
)
result.show()
# C1 Alice 1 2200.00
# C2 Bob   1  800.00
# C3 Carol 0    0.00

# Inspect intermediate outputs while practising.
# ranked_policies.orderBy("policy_id", "rn").show()
# latest_policies.orderBy("policy_id").show()
# active_policies.orderBy("policy_id").show()
# customer_summary.orderBy("customer_id").show()

Final · Q16–Q18 Time Pattern Comparison & Memory

exam recognition map

Start from the requested output shape. Merged ranges, a peak number, and elapsed duration are three different output shapes. Start from the requested output, then choose Q16, Q17, or Q18.

Pattern	Question asks for	Core transformation	Window / state to watch	Memory
Q16	Merged date / interval ranges	Find islands of overlapping intervals	previous MAX(end) over previous rows	SORT → PREV MAX → FLAG → GROUP → MERGE
Q17	Peak active count / minimum platforms	Intervals → signed boundary events	running SUM(delta)	EVENTS → NET → RUN → MAX
Q18	Time spent / active duration	Clean state log → pair boundaries	LAG for dedupe + LEAD for next boundary	ORDER → DEDUPE → PAIR → DURATION → SUM

Peak NUMBER?Use Q17. Convert starts and ends to +1 / -1 events, then running SUM and MAX.

Merged RANGES?Use Q16. previous_max_end decides whether the current interval joins the island or starts a new one.

DURATION?Use Q18. Reconstruct valid START → END periods first; only then subtract timestamps.

Two different cumulative SUM meanings: Q17 cumulative SUM of delta gives a changing count; Q16 cumulative SUM of new_group gives a group_id. Same Window technique, different state.

Important merge-interval trap: for Q16, use the maximum end among all previous rows. A simple LAG(end) can fail when one long interval contains several shorter intervals.

EXAM TRIGGER: MERGED RANGES → Q16 · PEAK NUMBER → Q17 · DURATION → Q18

SQL dialect: Spark SQL · Q16 uses DATE columns and inclusive date ranges. Q17–Q18 use non-NULL TIME values on the same day (HH:MM), with departure ≥ arrival. Q17 allows immediate platform reuse at departure. For overnight visits, supply full DATETIME values and use TIMESTAMPDIFF instead.

🔊 PRONUNCIATION QUICK REVIEW

Common Interview Words

Click any row to hear the word.

Word	Easy Pronunciation	Stress
enumerate	i-NU-ma-rate / 伊-努-么-rate	NU
analyze	AN-a-lyze / 艾-呢-laiz	AN
alias	A-lee-us / 诶-li-əs	A
truncate	TRUN-kate / 川-keɪt	TRUN
interval	IN-ter-val / 因-ter-vəl	IN
literal	LI-ter-al / 利-ter-rəl	LI
preceding	pre-CEED-ing / pri-SEE-ding	CEED
unbounded	un-BOUND-ed	BOUND

TAB 3 · BUS-TIME PYTHON SET

SQL ↔ Python · Bus Time Patterns

Focused practice for interval merging, maximum overlap, and arrival-to-departure duration. SQL stays on the left; plain Python stays on the right.

3 Python-focused problemsSQL ↔ Python4-space indentationIntervals + Arrival / Departure

Fast memory: Q16 asks for merged ranges; Q17 asks for a peak number; Q18 asks for duration.

Q16 · Merge Overlapping Date Ranges — Islands & Gaps

SORT → PREVIOUS MAX END → GROUP → MERGE

Pattern: SORT → PREVIOUS MAX END → GROUP → MERGE. This is the canonical interval-merge question used in Tab 2.

Problem

Given cmp_id, start_date, and end_date, merge overlapping intervals for each campaign and return one row per merged interval, including the number of calendar days in that merged range.

Sample Input · campaign

cmp_id	start_date	end_date
A1	2026-01-01	2026-01-05
A1	2026-01-04	2026-01-10
A1	2026-01-15	2026-01-17
A2	2026-02-01	2026-02-05
A2	2026-02-04	2026-02-08
A2	2026-02-07	2026-02-10

Sample Output

cmp_id	start_date	end_date	interval_days
A1	2026-01-01	2026-01-10	10
A1	2026-01-15	2026-01-17	3
A2	2026-02-01	2026-02-10	10

How to think: Group intervals by cmp_id, sort by start date, then compare the current start with the furthest end reached so far. In SQL that state is previous_max_end; in Python it is the end of the current merged range.

Key intermediate output · this is the table to understand

cmp_id	start_date	end_date	previous_max_end	new_group	group_id
A1	2026-01-01	2026-01-05	NULL	1	1
A1	2026-01-04	2026-01-10	2026-01-05	0	1
A1	2026-01-15	2026-01-17	2026-01-10	1	2
A2	2026-02-01	2026-02-05	NULL	1	1
A2	2026-02-04	2026-02-08	2026-02-05	0	1
A2	2026-02-07	2026-02-10	2026-02-08	0	1

The decisive test is start_date > previous_max_end. If true, start a new island; otherwise the interval belongs to the current island.

SQL · LEFT

SQL AnswerEditable · Save keeps your SQL in this browser

-- Q16: Merge overlapping date ranges.
WITH base AS (
    SELECT DISTINCT
        cmp_id,
        start_date,
        end_date
    FROM campaign
),
ordered AS (
    SELECT
        cmp_id,
        start_date,
        end_date,
        MAX(end_date) OVER (
            PARTITION BY cmp_id
            ORDER BY start_date, end_date
            ROWS BETWEEN UNBOUNDED PRECEDING AND 1 PRECEDING
        ) AS previous_max_end
    FROM base
),
marked AS (
    SELECT
        cmp_id,
        start_date,
        end_date,
        CASE
            WHEN previous_max_end IS NULL
                OR start_date > previous_max_end
            THEN 1
            ELSE 0
        END AS new_group
    FROM ordered
),
grouped AS (
    SELECT
        cmp_id,
        start_date,
        end_date,
        SUM(new_group) OVER (
            PARTITION BY cmp_id
            ORDER BY start_date, end_date
            ROWS BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW
        ) AS group_id
    FROM marked
),
merged AS (
    SELECT
        cmp_id,
        group_id,
        MIN(start_date) AS start_date,
        MAX(end_date) AS end_date
    FROM grouped
    GROUP BY cmp_id, group_id
)
SELECT
    cmp_id,
    start_date,
    end_date,
    DATEDIFF(end_date, start_date) + 1 AS interval_days
FROM merged
ORDER BY cmp_id, start_date;

PYTHON · RIGHT

from collections import defaultdict
from datetime import date

campaigns = [
    ("A1", "2026-01-01", "2026-01-05"),
    ("A1", "2026-01-04", "2026-01-10"),
    ("A1", "2026-01-15", "2026-01-17"),
    ("A2", "2026-02-01", "2026-02-05"),
    ("A2", "2026-02-04", "2026-02-08"),
    ("A2", "2026-02-07", "2026-02-10"),
]


def merge_date_ranges(rows):
    by_campaign = defaultdict(list)

    for cmp_id, start_date, end_date in rows:
        by_campaign[cmp_id].append((start_date, end_date))

    result = []

    for cmp_id, intervals in sorted(by_campaign.items()):
        merged = []

        for start_date, end_date in sorted(intervals):
            if not merged or start_date > merged[-1][1]:
                merged.append([start_date, end_date])
            else:
                merged[-1][1] = max(merged[-1][1], end_date)

        for start_date, end_date in merged:
            interval_days = (
                date.fromisoformat(end_date)
                - date.fromisoformat(start_date)
            ).days + 1

            result.append(
                (cmp_id, start_date, end_date, interval_days)
            )

    return result


print(merge_date_ranges(campaigns))

SQL ↔ Python mapping

SQL idea	Python equivalent
`PARTITION BY cmp_id`	dictionary keyed by `cmp_id`
`ORDER BY start_date`	`sorted(intervals)`
`previous_max_end`	current merged end: `merged[-1][1]`
`CASE new_group`	`if start > merged[-1][1]`
`MIN(start), MAX(end)`	append a new range or extend the last range
`DATEDIFF + 1`	`(end - start).days + 1`

Main difference

SQL creates an explicit group_id because rows are processed declaratively.
Python does not need a numeric group_id; the merged list itself is the current grouping state.
The algorithm is identical: compare the next start with the furthest end reached so far.

Q17 · Minimum Platforms — Maximum Concurrent Buses

EVENTS → NET → RUN → MAX

Pattern: EVENTS → NET → RUN → MAX. Read the key intermediate output before the code.

Problem

You are given the arrival and departure time of each bus. A platform can serve only one bus at a time. Return the minimum number of platforms required so that no bus has to wait.

Sample Input · bus

bus_id	arrival_time	departure_time
B1	09:00	09:30
B2	09:10	10:00
B3	09:20	09:50
B4	09:50	10:20
B5	10:00	10:30

Sample Output

min_platforms
3

How to think: Treat every bus as a half-open interval [arrival, departure). Arrival adds one active bus; departure removes one. First net equal timestamps, then scan the running count and keep the maximum.

Key intermediate output · this is the table to understand

event_time	net_delta	active_buses
09:00	+1	1
09:10	+1	2
09:20	+1	3
09:30	-1	2
09:50	0	2
10:00	0	2
10:20	-1	1
10:30	-1	0

At 09:50 and 10:00, one bus leaves and one bus arrives at the same time, so the net delta is 0. The peak active count is 3.

SQL · LEFT

SQL AnswerEditable · Save keeps your SQL in this browser

PYTHON · RIGHT

buses = [
    ("B1", "09:00", "09:30"),
    ("B2", "09:10", "10:00"),
    ("B3", "09:20", "09:50"),
    ("B4", "09:50", "10:20"),
    ("B5", "10:00", "10:30"),
]


def min_platforms(buses):
    # MEMORY: ARRIVAL +1 / DEPARTURE -1 -> NET -> RUN -> MAX
    events = []

    for _, arrival, departure in buses:
        events.append((arrival, 1))
        events.append((departure, -1))

    delta_by_time = {}

    for event_time, delta in events:
        delta_by_time[event_time] = (
            delta_by_time.get(event_time, 0) + delta
        )

    active = 0
    peak = 0

    for event_time in sorted(delta_by_time):
        active += delta_by_time[event_time]
        peak = max(peak, active)

    return peak


print(min_platforms(buses))  # 3

SQL → PYTHON MENTAL TRANSLATION Read left → middle → why

SQL concept	Python equivalent	How to think
Input table `bus`	List of tuples `buses = [...]`	A SQL table becomes an in-memory Python collection of records.
`UNION ALL` + `+1 / -1`	`events.append((time, delta))`	SQL emits two rows per interval; Python explicitly appends two event tuples.
`GROUP BY event_time` + `SUM(delta)`	Dictionary accumulation: `delta_by_time[t] += delta`	A Python dictionary plays the role of grouped state: key = group key, value = aggregate.
`ORDER BY event_time`	`sorted(delta_by_time)`	SQL ordering becomes an explicit sorted iteration in Python.
Running `SUM(...) OVER (...)`	`active += delta_by_time[event_time]`	The SQL window hides the loop; Python exposes the cumulative state variable.
`MAX(active_buses)`	`peak = max(peak, active)`	SQL aggregates the running column; Python updates the best value during the scan.

Core difference: SQL is set-based and declarative; Python must explicitly create rows/events, maintain grouped state in a dictionary, sort the keys, and update the running counter inside a loop.

Q18 · Total Dwell Time — Arrival-to-Departure Duration

PAIR → DURATION → SUM BY BUS

Pattern: PAIR → DURATION → SUM BY BUS. Read the key intermediate output before the code.

Problem

Each row is one paired bus station visit. Calculate the dwell duration of each visit and return total dwell minutes per bus.

Sample Input · bus_visit

bus_id	visit_id	arrival_time	departure_time
B1	VISIT1	09:00	09:30
B1	VISIT2	11:00	11:20
B2	VISIT1	09:10	10:00

Sample Output

bus_id	total_dwell_min
B1	50
B2	50

How to think: When arrival and departure are already paired, duration is simply departure - arrival. First compute dwell minutes per visit, then aggregate the durations by bus.

Key intermediate output · this is the table to understand

bus_id	visit_id	arrival_time	departure_time	dwell_min
B1	VISIT1	09:00	09:30	30
B1	VISIT2	11:00	11:20	20
B2	VISIT1	09:10	10:00	50

The key intermediate step is the visit-level duration table. After that, the final step is just a group-and-sum by bus_id.

SQL · LEFT

SQL AnswerEditable · Save keeps your SQL in this browser

PYTHON · RIGHT

from collections import defaultdict

visits = [
    ("B1", "VISIT1", "09:00", "09:30"),
    ("B1", "VISIT2", "11:00", "11:20"),
    ("B2", "VISIT1", "09:10", "10:00"),
]


def to_minutes(hhmm):
    hour, minute = map(int, hhmm.split(":"))
    return hour * 60 + minute


def total_dwell(visits):
    # MEMORY: PAIR -> DURATION -> SUM PER BUS
    totals = defaultdict(int)

    for bus_id, _, arrival, departure in visits:
        duration = (
            to_minutes(departure)
            - to_minutes(arrival)
        )
        totals[bus_id] += duration

    return sorted(totals.items())


print(total_dwell(visits))
# [('B1', 50), ('B2', 50)]

SQL → PYTHON MENTAL TRANSLATION Read left → middle → why

SQL concept	Python equivalent	How to think
Input table `bus_visit`	List of tuples `visits = [...]`	SQL reads rows from a table; Python iterates ordinary records.
`(TIME_TO_SEC(departure) - TIME_TO_SEC(arrival)) / 60`	`to_minutes(departure) - to_minutes(arrival)`	MySQL converts each TIME to seconds, subtracts, and divides by 60. Python converts each HH:MM string directly to minutes.
`GROUP BY bus_id`	`defaultdict(int)` keyed by `bus_id`	The dictionary key is the Python equivalent of the SQL grouping key.
`SUM(duration)`	`totals[bus_id] += duration`	The aggregate SUM becomes an explicit accumulator update.
`ORDER BY bus_id`	`sorted(totals.items())`	SQL sorts result rows; Python sorts the dictionary output.

Core difference: SQL combines row-level time arithmetic and grouped aggregation in one query. Python separates them into a conversion step plus a dictionary accumulation loop.

Three-question memory: Q17 asks for a peak number → events + running count + max. Q16 asks for merged ranges → sort + previous max end + merge. Q18 asks for duration → departure minus arrival + sum by bus. SQL is declarative; Python makes the same state changes explicit with lists, dictionaries, and loops.

a	b	c	d
x	100	c1	d1
x	200	c1	d2
x	150	c2	d1
x	300	c1	d3
x	250	c3	d2
x	120	c2	d1
x	180	c4	d4
x	90	c1	d2
y	50	c5	d5
y	80	c6	d5
z	30	c7	d6
z	40	c7	d7

a	b	c	d
x	100	c1	d1
x	200	c1	d2
x	150	c2	d1
x	300	c1	d3
x	250	c3	d2
x	120	c2	d1
x	180	c4	d4
x	90	c1	d2
y	50	c5	d5
y	80	c6	d5
z	30	c7	d6
z	40	c7	d7

a	b	c	d
x	100	c1	d1
x	200	c1	d2
x	150	c2	d1
x	300	c1	d3
x	250	c3	d2
x	120	c2	d1
x	180	c4	d4
x	90	c1	d2
y	50	c5	d5
y	80	c6	d5
z	30	c7	d6
z	40	c7	d7