SQL window functions: month-over-month balance growth and duplicate detection

-- Basic window function syntax
SELECT
    account_id,
    txn_month,
    avg_daily_balance,
    LAG(avg_daily_balance) OVER (
        PARTITION BY account_id
        ORDER BY txn_month
    ) AS prev_month_balance
FROM monthly_balances;

+------------+-----------+-------------------+--------------------+
| account_id | txn_month | avg_daily_balance | prev_month_balance |
+------------+-----------+-------------------+--------------------+
| ACC-001    | 2024-01   |         150000.00 |               NULL |
| ACC-001    | 2024-02   |         175000.00 |          150000.00 |
| ACC-001    | 2024-03   |         200000.00 |          175000.00 |
+------------+-----------+-------------------+--------------------+

-- Step 1: Compute monthly average daily balance
WITH monthly_avg AS (
    SELECT
        account_id,
        DATE_TRUNC('month', txn_date) AS txn_month,
        AVG(daily_balance) AS avg_balance
    FROM nbi_daily_balances
    WHERE txn_date >= CURRENT_DATE - INTERVAL '4 months'
    GROUP BY account_id, DATE_TRUNC('month', txn_date)
),
-- Step 2: Compare to previous month using LAG
with_prev AS (
    SELECT
        account_id,
        txn_month,
        avg_balance,
        LAG(avg_balance, 1) OVER (
            PARTITION BY account_id ORDER BY txn_month
        ) AS prev_1,
        LAG(avg_balance, 2) OVER (
            PARTITION BY account_id ORDER BY txn_month
        ) AS prev_2
    FROM monthly_avg
)
-- Step 3: Filter for 3 consecutive increases
SELECT account_id, txn_month, avg_balance
FROM with_prev
WHERE avg_balance > prev_1
  AND prev_1 > prev_2
ORDER BY account_id;

Result: Accounts with 3 months of continuous growth
+------------+-----------+-------------+
| account_id | txn_month | avg_balance |
+------------+-----------+-------------+
| ACC-10045  | 2024-03   |   250000.00 |
| ACC-10098  | 2024-03   |   180000.00 |
+------------+-----------+-------------+

-- Detect duplicate transactions
WITH numbered AS (
    SELECT
        txn_id,
        account_id,
        amount,
        txn_time,
        ROW_NUMBER() OVER (
            PARTITION BY account_id, amount, txn_time
            ORDER BY txn_id  -- earliest ID = original
        ) AS rn
    FROM nbi_transactions
)
-- List duplicates (rn > 1)
SELECT txn_id, account_id, amount, txn_time
FROM numbered
WHERE rn > 1
ORDER BY account_id, txn_time;

Duplicate transactions found:
+--------+------------+-----------+---------------------+
| txn_id | account_id |    amount |            txn_time |
+--------+------------+-----------+---------------------+
| TXN-55 | ACC-10045  | 25000.00  | 2024-03-15 14:30:00 |
| TXN-89 | ACC-10098  |  5000.00  | 2024-03-20 09:15:00 |
+--------+------------+-----------+---------------------+

-- Keep only the original transaction (deduplicate)
WITH numbered AS (
    SELECT *,
        ROW_NUMBER() OVER (
            PARTITION BY account_id, amount, txn_time
            ORDER BY txn_id
        ) AS rn
    FROM nbi_transactions
)
SELECT txn_id, account_id, amount, txn_time
FROM numbered
WHERE rn = 1;  -- only the first occurrence

import org.apache.spark.sql.expressions.Window;
import org.apache.spark.sql.expressions.WindowSpec;
import static org.apache.spark.sql.functions.*;

// DataFrame API equivalent of LAG for month-over-month
WindowSpec monthWindow = Window
    .partitionBy("accountId")
    .orderBy("txnMonth");

Dataset<Row> withPrev = monthlyAvg
    .withColumn("prevBalance", lag("avgBalance", 1).over(monthWindow))
    .withColumn("prevBalance2", lag("avgBalance", 2).over(monthWindow));

Dataset<Row> growing = withPrev
    .filter(col("avgBalance").gt(col("prevBalance")))
    .filter(col("prevBalance").gt(col("prevBalance2")));

// DataFrame API equivalent of ROW_NUMBER for dedup
WindowSpec dedupWindow = Window
    .partitionBy("accountId", "amount", "txnTime")
    .orderBy("txnId");

Dataset<Row> deduped = transactions
    .withColumn("rn", row_number().over(dedupWindow))
    .filter(col("rn").equalTo(1))
    .drop("rn");

-- Handle null from LAG (first month has no previous)
SELECT
    account_id,
    txn_month,
    avg_balance,
    COALESCE(
        LAG(avg_balance) OVER (PARTITION BY account_id ORDER BY txn_month),
        avg_balance  -- default: no growth if no previous month
    ) AS prev_balance
FROM monthly_avg;

-- Distinguish true dupes from legitimate repeats
-- Include transaction_ref to avoid false positives
ROW_NUMBER() OVER (
    PARTITION BY account_id, amount, txn_time, merchant_id
    ORDER BY txn_id
)
-- Two separate coffees at the same merchant at the same time
-- are still different if they have different transaction_refs

-- Performance: partition table by month, index on account_id
CREATE INDEX idx_balance_acct_month
    ON nbi_daily_balances (account_id, txn_date);

-- The optimizer can scan only the relevant months
SELECT *
FROM nbi_daily_balances
WHERE txn_date >= CURRENT_DATE - INTERVAL '4 months';

-- WRONG: ORDER BY random column, "original" is unpredictable
ROW_NUMBER() OVER (
    PARTITION BY account_id, amount, txn_time
    ORDER BY RANDOM()  -- which row is "first" is random!
)

-- CORRECT: ORDER BY a stable, meaningful column
ROW_NUMBER() OVER (
    PARTITION BY account_id, amount, txn_time
    ORDER BY txn_id  -- earliest ID is the original
)