SQL Interview Questions for Data Science

┌──────────────────────────────────────────────────────────────────────┐
│                   SQL QUERY EXECUTION ORDER                          │
├──────────────────────────────────────────────────────────────────────┤
│                                                                      │
│  1. FROM            ➜  Read table(s)                                │
│  2. JOIN            ➜  Combine tables                               │
│  3. WHERE ⚡        ➜  Filter ROWS (before grouping)                │
│  4. GROUP BY        ➜  Create groups                                │
│  5. HAVING ⚡       ➜  Filter GROUPS (after aggregation)            │
│  6. SELECT          ➜  Choose columns                               │
│  7. DISTINCT        ➜  Remove duplicates                            │
│  8. ORDER BY        ➜  Sort results                                 │
│  9. LIMIT/OFFSET    ➜  Paginate                                     │
│                                                                      │
│  WHERE:  Filters 1M rows → 500K rows → Then group                  │
│  HAVING: Group all 1M rows → Filter groups                         │
│                                                                      │
└──────────────────────────────────────────────────────────────────────┘

import pandas as pd
import numpy as np
from typing import List, Dict, Tuple
from dataclasses import dataclass
import time

@dataclass
class QueryMetrics:
    """Track query performance metrics."""
    rows_scanned: int
    rows_filtered: int
    groups_created: int
    groups_filtered: int
    execution_time_ms: float
    memory_usage_mb: float

class SQLExecutionEngine:
    """Simulates SQL WHERE vs HAVING execution."""

    def __init__(self, data: pd.DataFrame):
        """Initialize with sample dataset."""
        self.data = data
        self.original_size = len(data)

    def query_with_where(
        self, 
        where_condition: str,
        group_by_col: str,
        having_condition: str = None
    ) -> Tuple[pd.DataFrame, QueryMetrics]:
        """
        Execute query with WHERE clause (efficient approach).

        Execution order: WHERE → GROUP BY → HAVING
        """
        start_time = time.time()

        # Step 1: WHERE filters rows FIRST (reduces dataset)
        filtered_data = self.data.query(where_condition)
        rows_after_where = len(filtered_data)

        # Step 2: GROUP BY on filtered data
        grouped = filtered_data.groupby(group_by_col).agg({
            'sales': 'sum',
            'employee_id': 'count'
        }).reset_index()
        grouped.columns = [group_by_col, 'total_sales', 'emp_count']
        groups_created = len(grouped)

        # Step 3: HAVING filters groups (if provided)
        if having_condition:
            result = grouped.query(having_condition)
            groups_after_having = len(result)
        else:
            result = grouped
            groups_after_having = groups_created

        execution_time = (time.time() - start_time) * 1000

        metrics = QueryMetrics(
            rows_scanned=self.original_size,
            rows_filtered=rows_after_where,
            groups_created=groups_created,
            groups_filtered=groups_after_having,
            execution_time_ms=execution_time,
            memory_usage_mb=result.memory_usage(deep=True).sum() / 1024 / 1024
        )

        return result, metrics

    def query_without_where(
        self,
        group_by_col: str,
        having_condition: str
    ) -> Tuple[pd.DataFrame, QueryMetrics]:
        """
        Execute query with only HAVING clause (inefficient approach).

        Execution order: GROUP BY all data → HAVING
        """
        start_time = time.time()

        # Step 1: GROUP BY on ALL data (no WHERE filtering)
        grouped = self.data.groupby(group_by_col).agg({
            'sales': 'sum',
            'employee_id': 'count'
        }).reset_index()
        grouped.columns = [group_by_col, 'total_sales', 'emp_count']
        groups_created = len(grouped)

        # Step 2: HAVING filters groups
        result = grouped.query(having_condition)
        groups_after_having = len(result)

        execution_time = (time.time() - start_time) * 1000

        metrics = QueryMetrics(
            rows_scanned=self.original_size,
            rows_filtered=self.original_size,  # No WHERE filtering
            groups_created=groups_created,
            groups_filtered=groups_after_having,
            execution_time_ms=execution_time,
            memory_usage_mb=result.memory_usage(deep=True).sum() / 1024 / 1024
        )

        return result, metrics

# ============================================================================
# EXAMPLE 1: GOOGLE - USER ANALYTICS QUERY OPTIMIZATION
# ============================================================================
print("="*70)
print("GOOGLE - USER ANALYTICS: WHERE vs HAVING Performance")
print("="*70)

# Simulate Google user activity data (10M records)
np.random.seed(42)
n_records = 10_000_000

google_data = pd.DataFrame({
    'user_id': np.random.randint(1, 1000000, n_records),
    'country': np.random.choice(['US', 'UK', 'IN', 'CA', 'DE'], n_records),
    'sales': np.random.randint(10, 10000, n_records),
    'employee_id': np.random.randint(1, 100000, n_records),
    'is_premium': np.random.choice([0, 1], n_records, p=[0.7, 0.3])
})

engine = SQLExecutionEngine(google_data)

# Query: Find countries with >100 premium users and total sales > $10M

# Method 1: EFFICIENT - Filter with WHERE first
result1, metrics1 = engine.query_with_where(
    where_condition='is_premium == 1',  # Filter 3M → 10M rows
    group_by_col='country',
    having_condition='total_sales > 10000000'
)

print(f"\n✅ EFFICIENT APPROACH (WHERE → GROUP BY → HAVING):")
print(f"   Rows scanned:     {metrics1.rows_scanned:,}")
print(f"   Rows filtered:    {metrics1.rows_filtered:,} ({metrics1.rows_filtered/metrics1.rows_scanned*100:.1f}%)")
print(f"   Groups created:   {metrics1.groups_created:,}")
print(f"   Groups filtered:  {metrics1.groups_filtered:,}")
print(f"   Execution time:   {metrics1.execution_time_ms:.2f}ms")
print(f"   Memory usage:     {metrics1.memory_usage_mb:.2f}MB")

# Method 2: INEFFICIENT - Group all data, then filter
result2, metrics2 = engine.query_without_where(
    group_by_col='country',
    having_condition='emp_count > 100000'  # Would need WHERE for sales
)

print(f"\n❌ INEFFICIENT APPROACH (GROUP BY all → HAVING only):")
print(f"   Rows scanned:     {metrics2.rows_scanned:,}")
print(f"   Rows filtered:    {metrics2.rows_filtered:,} (100.0%)")
print(f"   Groups created:   {metrics2.groups_created:,}")
print(f"   Groups filtered:  {metrics2.groups_filtered:,}")
print(f"   Execution time:   {metrics2.execution_time_ms:.2f}ms")
print(f"   Memory usage:     {metrics2.memory_usage_mb:.2f}MB")

speedup = metrics2.execution_time_ms / metrics1.execution_time_ms
print(f"\n⚡ Performance improvement: {speedup:.2f}x faster with WHERE clause!")
print(f"   Google saved {(metrics2.execution_time_ms - metrics1.execution_time_ms):.2f}ms per query")
print(f"   On 1M queries/day: {(speedup - 1) * 100:.1f}% cost reduction")

# ============================================================================
# EXAMPLE 2: META - AD CAMPAIGN ANALYSIS
# ============================================================================
print("\n" + "="*70)
print("META - AD CAMPAIGN ANALYSIS: Complex WHERE + HAVING")
print("="*70)

meta_campaigns = pd.DataFrame({
    'campaign_id': range(1, 101),
    'advertiser_id': np.random.randint(1, 21, 100),
    'impressions': np.random.randint(10000, 1000000, 100),
    'clicks': np.random.randint(100, 50000, 100),
    'spend': np.random.uniform(100, 50000, 100),
    'conversions': np.random.randint(0, 1000, 100),
    'is_active': np.random.choice([0, 1], 100, p=[0.3, 0.7])
})

meta_campaigns['ctr'] = (meta_campaigns['clicks'] / meta_campaigns['impressions']) * 100
meta_campaigns['cpc'] = meta_campaigns['spend'] / meta_campaigns['clicks']

# SQL Query Equivalent:
sql_query = """
SELECT advertiser_id, 
       COUNT(*) as campaign_count,
       SUM(spend) as total_spend,
       AVG(ctr) as avg_ctr
FROM campaigns
WHERE is_active = 1           -- WHERE: Filter rows
  AND ctr > 2.0               -- WHERE: Filter rows
  AND cpc < 5.0               -- WHERE: Filter rows
GROUP BY advertiser_id
HAVING SUM(spend) > 100000    -- HAVING: Filter groups
   AND COUNT(*) >= 5;         -- HAVING: Filter groups
"""

print(f"\nSQL Query:\n{sql_query}")

# Execute with WHERE filtering
active_campaigns = meta_campaigns[
    (meta_campaigns['is_active'] == 1) &
    (meta_campaigns['ctr'] > 2.0) &
    (meta_campaigns['cpc'] < 5.0)
]

advertiser_stats = active_campaigns.groupby('advertiser_id').agg({
    'campaign_id': 'count',
    'spend': 'sum',
    'ctr': 'mean'
}).reset_index()
advertiser_stats.columns = ['advertiser_id', 'campaign_count', 'total_spend', 'avg_ctr']

# Apply HAVING conditions
result = advertiser_stats[
    (advertiser_stats['total_spend'] > 100000) &
    (advertiser_stats['campaign_count'] >= 5)
]

print(f"\nResults:")
print(f"   Total campaigns: {len(meta_campaigns):,}")
print(f"   After WHERE filtering: {len(active_campaigns):,}")
print(f"   Groups created: {len(advertiser_stats):,}")
print(f"   After HAVING filtering: {len(result):,}")
print(f"\n   Meta identified {len(result)} high-value advertisers")
print(f"   Combined spend: ${result['total_spend'].sum():,.2f}")
print(f"   Average CTR: {result['avg_ctr'].mean():.3f}%")

print("\n" + "="*70)
print("Key Takeaway: WHERE reduces 70% of data before aggregation!")
print("="*70)

-- ====================================================================
-- PATTERN 1: E-commerce - High-value customers
-- ====================================================================
SELECT customer_id, 
       COUNT(*) as order_count,
       SUM(order_total) as total_spent,
       AVG(order_total) as avg_order_value
FROM orders
WHERE order_date >= '2024-01-01'        -- WHERE: Date filter
  AND status = 'completed'              -- WHERE: Status filter
  AND payment_method != 'cancelled'     -- WHERE: Exclude refunds
GROUP BY customer_id
HAVING SUM(order_total) > 10000         -- HAVING: High spenders
   AND COUNT(*) >= 10;                  -- HAVING: Frequent buyers

-- ====================================================================
-- PATTERN 2: SaaS - Active user cohorts
-- ====================================================================
SELECT cohort_month, 
       COUNT(DISTINCT user_id) as active_users,
       SUM(revenue) as total_revenue,
       AVG(session_count) as avg_sessions
FROM user_activity
WHERE last_active >= CURRENT_DATE - INTERVAL '30 days'  -- WHERE: Recent activity
  AND subscription_status = 'active'                     -- WHERE: Paid users
  AND user_tier IN ('premium', 'enterprise')             -- WHERE: Exclude free
GROUP BY cohort_month
HAVING COUNT(DISTINCT user_id) >= 100   -- HAVING: Significant cohort size
   AND AVG(session_count) > 20;         -- HAVING: High engagement

-- ====================================================================
-- PATTERN 3: Finance - Fraud detection
-- ====================================================================
SELECT account_id,
       COUNT(*) as transaction_count,
       SUM(amount) as total_amount,
       MAX(amount) as max_transaction
FROM transactions
WHERE transaction_date >= CURRENT_DATE - INTERVAL '24 hours'  -- WHERE: Recent
  AND transaction_type = 'withdrawal'                          -- WHERE: Type
  AND location_country != home_country                         -- WHERE: Foreign
  AND risk_score > 50                                          -- WHERE: Suspicious
GROUP BY account_id
HAVING COUNT(*) >= 5                    -- HAVING: High frequency
   AND SUM(amount) > 50000              -- HAVING: Large total
   AND MAX(amount) > 10000;             -- HAVING: Large single transaction

-- ❌ ANTI-PATTERN 1: Using HAVING for row-level filtering
SELECT department, employee_name, salary
FROM employees
GROUP BY department, employee_name, salary
HAVING salary > 100000;  -- WRONG! Should use WHERE

-- ✅ CORRECT: Use WHERE for row filtering
SELECT department, employee_name, salary
FROM employees
WHERE salary > 100000;

-- ❌ ANTI-PATTERN 2: Filtering all data after grouping
SELECT city, COUNT(*) as user_count
FROM users
GROUP BY city
HAVING city IN ('New York', 'Los Angeles', 'Chicago');  -- Inefficient

-- ✅ CORRECT: Filter rows before grouping
SELECT city, COUNT(*) as user_count
FROM users
WHERE city IN ('New York', 'Los Angeles', 'Chicago')  -- Uses index
GROUP BY city;

-- ❌ ANTI-PATTERN 3: Complex calculations in HAVING
SELECT product_id, 
       SUM(quantity * price) as revenue
FROM sales
GROUP BY product_id
HAVING SUM(quantity * price) / COUNT(*) > 1000;  -- Recalculates aggregate

-- ✅ CORRECT: Use subquery or CTE
WITH product_revenue AS (
    SELECT product_id, 
           SUM(quantity * price) as revenue,
           COUNT(*) as order_count
    FROM sales
    GROUP BY product_id
)
SELECT * 
FROM product_revenue
WHERE revenue / order_count > 1000;  -- Simple division

┌──────────────────────────────────────────────────────────────────────┐
│                         JOIN TYPES VISUAL                            │
├──────────────────────────────────────────────────────────────────────┤
│                                                                      │
│  TABLE A (employees)          TABLE B (departments)                 │
│  ┌─────────────────┐          ┌─────────────────┐                 │
│  │  1  Alice  10   │          │  10  Engineering│                 │
│  │  2  Bob    20   │          │  20  Marketing  │                 │
│  │  3  Carol  NULL │          │  30  HR         │                 │
│  └─────────────────┘          └─────────────────┘                 │
│                                                                      │
│  INNER JOIN: Intersection only (matching records)                   │
│  ┌─────────────────────────────────────────┐                       │
│  │ Alice  →  Engineering                   │                       │
│  │ Bob    →  Marketing                     │                       │
│  └─────────────────────────────────────────┘                       │
│  Result: 2 rows (Carol not included, HR not included)              │
│                                                                      │
│  LEFT (OUTER) JOIN: All from A + matching from B                    │
│  ┌─────────────────────────────────────────┐                       │
│  │ Alice  →  Engineering                   │                       │
│  │ Bob    →  Marketing                     │                       │
│  │ Carol  →  NULL                          │                       │
│  └─────────────────────────────────────────┘                       │
│  Result: 3 rows (all employees, Carol has no dept)                 │
│                                                                      │
│  RIGHT (OUTER) JOIN: All from B + matching from A                   │
│  ┌─────────────────────────────────────────┐                       │
│  │ Alice  →  Engineering                   │                       │
│  │ Bob    →  Marketing                     │                       │
│  │ NULL   →  HR                            │                       │
│  └─────────────────────────────────────────┘                       │
│  Result: 3 rows (all depts, HR has no employee)                    │
│                                                                      │
│  FULL OUTER JOIN: All from A + All from B                           │
│  ┌─────────────────────────────────────────┐                       │
│  │ Alice  →  Engineering                   │                       │
│  │ Bob    →  Marketing                     │                       │
│  │ Carol  →  NULL                          │                       │
│  │ NULL   →  HR                            │                       │
│  └─────────────────────────────────────────┘                       │
│  Result: 4 rows (everyone and everything)                          │
│                                                                      │
│  CROSS JOIN: Cartesian product (A × B)                              │
│  ┌─────────────────────────────────────────┐                       │
│  │ Alice  ×  Engineering, Marketing, HR    │                       │
│  │ Bob    ×  Engineering, Marketing, HR    │                       │
│  │ Carol  ×  Engineering, Marketing, HR    │                       │
│  └─────────────────────────────────────────┘                       │
│  Result: 9 rows (3 × 3 = every combination)                        │
│                                                                      │
└──────────────────────────────────────────────────────────────────────┘

import pandas as pd
import numpy as np
from typing import List, Dict, Tuple, Optional
from dataclasses import dataclass
from enum import Enum
import time

class JoinType(Enum):
    """Enum for different JOIN types."""
    INNER = "inner"
    LEFT = "left"
    RIGHT = "right"
    OUTER = "outer"
    CROSS = "cross"

@dataclass
class JoinMetrics:
    """Track JOIN performance metrics."""
    left_rows: int
    right_rows: int
    result_rows: int
    null_count: int
    execution_time_ms: float
    join_selectivity: float  # result_rows / (left_rows * right_rows)

class JoinEngine:
    """Simulates different SQL JOIN operations."""

    def __init__(self, left_df: pd.DataFrame, right_df: pd.DataFrame):
        """Initialize with two dataframes."""
        self.left_df = left_df
        self.right_df = right_df

    def execute_join(
        self,
        join_type: JoinType,
        left_on: str,
        right_on: str,
        suffixes: Tuple[str, str] = ('_left', '_right')
    ) -> Tuple[pd.DataFrame, JoinMetrics]:
        """
        Execute specified JOIN type and return metrics.
        """
        start_time = time.time()
        left_rows = len(self.left_df)
        right_rows = len(self.right_df)

        # Execute JOIN based on type
        if join_type == JoinType.INNER:
            result = pd.merge(
                self.left_df, self.right_df,
                left_on=left_on, right_on=right_on,
                how='inner', suffixes=suffixes
            )
        elif join_type == JoinType.LEFT:
            result = pd.merge(
                self.left_df, self.right_df,
                left_on=left_on, right_on=right_on,
                how='left', suffixes=suffixes
            )
        elif join_type == JoinType.RIGHT:
            result = pd.merge(
                self.left_df, self.right_df,
                left_on=left_on, right_on=right_on,
                how='right', suffixes=suffixes
            )
        elif join_type == JoinType.OUTER:
            result = pd.merge(
                self.left_df, self.right_df,
                left_on=left_on, right_on=right_on,
                how='outer', suffixes=suffixes
            )
        elif join_type == JoinType.CROSS:
            result = self.left_df.merge(self.right_df, how='cross')
        else:
            raise ValueError(f"Unsupported join type: {join_type}")

        execution_time = (time.time() - start_time) * 1000
        result_rows = len(result)
        null_count = result.isnull().sum().sum()

        # Calculate join selectivity (0-1, higher = more matches)
        cartesian_product = left_rows * right_rows if right_rows > 0 else 1
        selectivity = result_rows / cartesian_product if cartesian_product > 0 else 0

        metrics = JoinMetrics(
            left_rows=left_rows,
            right_rows=right_rows,
            result_rows=result_rows,
            null_count=null_count,
            execution_time_ms=execution_time,
            join_selectivity=selectivity
        )

        return result, metrics

    def anti_join(self, left_on: str, right_on: str) -> pd.DataFrame:
        """
        Execute ANTI JOIN (LEFT JOIN + NULL filter).

        Returns rows from left that have NO match in right.
        """
        merged = pd.merge(
            self.left_df, self.right_df,
            left_on=left_on, right_on=right_on,
            how='left', indicator=True
        )
        return merged[merged['_merge'] == 'left_only'].drop('_merge', axis=1)

    def semi_join(self, left_on: str, right_on: str) -> pd.DataFrame:
        """
        Execute SEMI JOIN (EXISTS pattern).

        Returns rows from left that have a match in right.
        """
        return self.left_df[
            self.left_df[left_on].isin(self.right_df[right_on])
        ]

# ============================================================================
# EXAMPLE 1: NETFLIX - USER VIEWING HISTORY WITH METADATA
# ============================================================================
print("="*70)
print("NETFLIX - JOIN VIEWING HISTORY WITH CONTENT METADATA")
print("="*70)

# Netflix viewing events (10M records/day)
np.random.seed(42)
viewing_events = pd.DataFrame({
    'user_id': np.random.randint(1, 1000000, 100000),
    'content_id': np.random.randint(1, 10001, 100000),
    'watch_duration_min': np.random.randint(1, 180, 100000),
    'device': np.random.choice(['TV', 'Mobile', 'Web'], 100000),
    'timestamp': pd.date_range('2024-01-01', periods=100000, freq='10S')
})

# Content catalog (10K titles)
content_catalog = pd.DataFrame({
    'content_id': range(1, 10001),
    'title': [f'Title_{i}' for i in range(1, 10001)],
    'genre': np.random.choice(['Action', 'Drama', 'Comedy', 'Sci-Fi'], 10000),
    'release_year': np.random.randint(2010, 2025, 10000),
    'rating': np.random.uniform(6.0, 9.5, 10000)
})

engine = JoinEngine(viewing_events, content_catalog)

# INNER JOIN: Only matching content (production scenario)
result_inner, metrics_inner = engine.execute_join(
    JoinType.INNER, 'content_id', 'content_id'
)

print(f"\n✅ INNER JOIN (Netflix Production Query):")
print(f"   Viewing events:   {metrics_inner.left_rows:,}")
print(f"   Content catalog:  {metrics_inner.right_rows:,}")
print(f"   Result rows:      {metrics_inner.result_rows:,}")
print(f"   Null values:      {metrics_inner.null_count}")
print(f"   Execution time:   {metrics_inner.execution_time_ms:.2f}ms")
print(f"   Join selectivity: {metrics_inner.join_selectivity:.6f}")
print(f"\n   Netflix processes {metrics_inner.result_rows:,} enriched events/batch")
print(f"   Average watch time: {result_inner['watch_duration_min'].mean():.1f} minutes")

# LEFT JOIN: Find events with missing content metadata (data quality check)
result_left, metrics_left = engine.execute_join(
    JoinType.LEFT, 'content_id', 'content_id'
)
missing_metadata = result_left[result_left['title'].isnull()]

print(f"\n🔍 LEFT JOIN (Data Quality Check):")
print(f"   Total events:           {metrics_left.result_rows:,}")
print(f"   Missing metadata:       {len(missing_metadata):,}")
print(f"   Data quality:           {(1 - len(missing_metadata)/metrics_left.result_rows)*100:.2f}%")

# ============================================================================
# EXAMPLE 2: SPOTIFY - SELF JOIN FOR SONG RECOMMENDATIONS
# ============================================================================
print("\n" + "="*70)
print("SPOTIFY - SELF JOIN FOR COLLABORATIVE FILTERING")
print("="*70)

# User listening history
user_listens = pd.DataFrame({
    'user_id': [1, 1, 1, 2, 2, 2, 3, 3, 3, 4, 4, 4],
    'song_id': [101, 102, 103, 101, 104, 105, 102, 103, 106, 101, 102, 107],
    'play_count': np.random.randint(1, 100, 12)
})

# Self-join to find users with similar tastes
similar_users = pd.merge(
    user_listens, user_listens,
    on='song_id', suffixes=('_user1', '_user2')
)
# Filter: different users, same songs
similar_users = similar_users[
    similar_users['user_id_user1'] < similar_users['user_id_user2']
]

similarity_scores = similar_users.groupby(
    ['user_id_user1', 'user_id_user2']
).size().reset_index(name='common_songs')

print(f"\nSelf-JOIN results:")
print(f"   Total listening events: {len(user_listens)}")
print(f"   User pairs found:       {len(similarity_scores)}")
print(f"\nTop similar user pairs:")
print(similarity_scores.sort_values('common_songs', ascending=False).head())
print(f"\n   Spotify uses this pattern for 'Users who liked this also liked...'")

# ============================================================================
# EXAMPLE 3: UBER - CROSS JOIN FOR DRIVER-RIDER MATCHING (Limited)
# ============================================================================
print("\n" + "="*70)
print("UBER - LIMITED CROSS JOIN FOR DRIVER MATCHING")
print("="*70)

# Available drivers in area
drivers = pd.DataFrame({
    'driver_id': [1, 2, 3],
    'lat': [37.7749, 37.7750, 37.7751],
    'lon': [-122.4194, -122.4195, -122.4196],
    'rating': [4.8, 4.9, 4.7]
})

# Ride requests
riders = pd.DataFrame({
    'rider_id': [101, 102],
    'pickup_lat': [37.7749, 37.7750],
    'pickup_lon': [-122.4194, -122.4195],
    'requested_at': pd.Timestamp('2024-01-01 10:00:00')
})

# CROSS JOIN to generate all driver-rider combinations
all_matches = drivers.merge(riders, how='cross')

# Calculate distance (simplified Euclidean)
all_matches['distance'] = np.sqrt(
    (all_matches['lat'] - all_matches['pickup_lat'])**2 +
    (all_matches['lon'] - all_matches['pickup_lon'])**2
)

# Find best match for each rider (closest driver with high rating)
best_matches = all_matches.loc[
    all_matches.groupby('rider_id')['distance'].idxmin()
]

print(f"\nCROSS JOIN for matching:")
print(f"   Drivers available: {len(drivers)}")
print(f"   Ride requests:     {len(riders)}")
print(f"   Combinations:      {len(all_matches)} ({len(drivers)} × {len(riders)})")
print(f"\nBest matches:")
print(best_matches[['rider_id', 'driver_id', 'distance', 'rating']])
print(f"\n   ⚠️  Uber limits search radius to prevent O(n²) explosion")
print(f"   Only drivers within 0.5 miles are considered")

# ============================================================================
# EXAMPLE 4: ANTI-JOIN AND SEMI-JOIN PATTERNS
# ============================================================================
print("\n" + "="*70)
print("GOOGLE ADS - ANTI-JOIN FOR NEGATIVE TARGETING")
print("="*70)

# All potential ad impressions
ad_opportunities = pd.DataFrame({
    'user_id': [1, 2, 3, 4, 5],
    'page_url': ['url1', 'url2', 'url3', 'url4', 'url5']
})

# Users who already converted (should NOT see ad)
converted_users = pd.DataFrame({
    'user_id': [2, 4],
    'conversion_date': pd.Timestamp('2024-01-01')
})

# ANTI-JOIN: Find users who haven't converted
engine_ad = JoinEngine(ad_opportunities, converted_users)
non_converted = engine_ad.anti_join('user_id', 'user_id')

print(f"\nAnti-JOIN results:")
print(f"   Potential impressions:  {len(ad_opportunities)}")
print(f"   Already converted:      {len(converted_users)}")
print(f"   Should show ad:         {len(non_converted)}")
print(f"   Users: {non_converted['user_id'].tolist()}")
print(f"\n   Google saved {len(converted_users)/len(ad_opportunities)*100:.1f}% of ad spend")

print("\n" + "="*70)
print("Key Takeaway: JOIN choice impacts both correctness and performance!")
print("="*70)

-- ====================================================================
-- NETFLIX: Enrich viewing events with content metadata
-- ====================================================================
SELECT 
    v.user_id,
    v.watch_duration_min,
    c.title,
    c.genre,
    c.rating
FROM viewing_events v
INNER JOIN content_catalog c ON v.content_id = c.content_id
WHERE v.timestamp >= CURRENT_DATE - INTERVAL '7 days'
  AND c.rating >= 8.0;

-- Performance: 10M rows/sec with indexed content_id

-- ====================================================================
-- SPOTIFY: Find similar users (collaborative filtering)
-- ====================================================================
SELECT 
    u1.user_id as user1,
    u2.user_id as user2,
    COUNT(DISTINCT u1.song_id) as common_songs
FROM user_listens u1
INNER JOIN user_listens u2 
    ON u1.song_id = u2.song_id
    AND u1.user_id < u2.user_id  -- Avoid duplicate pairs
GROUP BY u1.user_id, u2.user_id
HAVING COUNT(DISTINCT u1.song_id) >= 10;  -- At least 10 common songs

-- Spotify uses this for "Fans also like" feature

-- ====================================================================
-- AIRBNB: Find unbooked listings (LEFT JOIN + NULL filter)
-- ====================================================================
SELECT l.listing_id, l.title, l.price_per_night
FROM listings l
LEFT JOIN bookings b 
    ON l.listing_id = b.listing_id
    AND b.checkin_date BETWEEN '2024-12-01' AND '2024-12-31'
WHERE b.booking_id IS NULL  -- No bookings in December
  AND l.is_active = true;

-- Airbnb sends promotional emails to these listings

-- ====================================================================
-- UBER: Driver-rider matching (CROSS JOIN with constraints)
-- ====================================================================
SELECT 
    d.driver_id,
    r.rider_id,
    ST_Distance(d.location, r.pickup_location) as distance_miles,
    d.rating
FROM available_drivers d
CROSS JOIN ride_requests r
WHERE ST_Distance(d.location, r.pickup_location) < 0.5  -- Within 0.5 miles
  AND d.rating >= 4.5
  AND r.created_at >= CURRENT_TIMESTAMP - INTERVAL '5 minutes'
ORDER BY distance_miles, d.rating DESC;

-- Uber limits CROSS JOIN with geo constraints to prevent explosion

-- ====================================================================
-- META: Multi-way JOIN for ad targeting
-- ====================================================================
SELECT 
    u.user_id,
    u.demographics,
    i.interest_category,
    a.ad_creative_id,
    a.bid_amount
FROM users u
INNER JOIN user_interests i ON u.user_id = i.user_id
INNER JOIN ad_campaigns a 
    ON i.interest_category = a.target_category
    AND u.age_group = a.target_age
LEFT JOIN ad_impressions ai
    ON u.user_id = ai.user_id
    AND a.ad_creative_id = ai.ad_creative_id
    AND ai.shown_at >= CURRENT_DATE - INTERVAL '7 days'
WHERE ai.impression_id IS NULL  -- Haven't seen this ad recently
  AND a.status = 'active'
LIMIT 1000;

-- Meta's 3-table JOIN for ad delivery

┌──────────────────────────────────────────────────────────────────────┐
│                    WINDOW FUNCTION ANATOMY                           │
├──────────────────────────────────────────────────────────────────────┤
│                                                                      │
│  function_name() OVER (                                             │
│      PARTITION BY col1, col2    ← Divide into groups (like GROUP BY)│
│      ORDER BY col3              ← Order within each partition       │
│      ROWS/RANGE frame_spec      ← Define sliding window             │
│  )                                                                   │
│                                                                      │
│  ┌────────────────────────────────────────────────────────────────┐ │
│  │ PARTITION BY vs GROUP BY                                      │ │
│  ├────────────────────────────────────────────────────────────────┤ │
│  │ GROUP BY:                                                       │ │
│  │   Input: 1000 rows  →  Output: 10 groups (collapses)          │ │
│  │   Returns: Aggregated values only                              │ │
│  │                                                                │ │
│  │ PARTITION BY:                                                   │ │
│  │   Input: 1000 rows  →  Output: 1000 rows (preserves)          │ │
│  │   Returns: Original + calculated columns                       │ │
│  └────────────────────────────────────────────────────────────────┘ │
│                                                                      │
│  Frame Specifications (Sliding Window):                             │
│  ┌────────────────────────────────────────────────────────────────┐ │
│  │ ROWS BETWEEN 2 PRECEDING AND CURRENT ROW                       │ │
│  │                                                                │ │
│  │     [Row N-2]  [Row N-1]  [Row N] ← Current                   │ │
│  │      ↑─────────────────────────↑                              │ │
│  │      └──── Window for Row N ───┘                              │ │
│  │                                                                │ │
│  │ Common frame specs:                                            │ │
│  │   • UNBOUNDED PRECEDING: From start to current                │ │
│  │   • N PRECEDING: N rows before current                        │ │
│  │   • CURRENT ROW: Just current row                             │ │
│  │   • N FOLLOWING: N rows after current                         │ │
│  │   • UNBOUNDED FOLLOWING: Current to end                       │ │
│  └────────────────────────────────────────────────────────────────┘ │
│                                                                      │
└──────────────────────────────────────────────────────────────────────┘

import pandas as pd
import numpy as np
from typing import List, Dict, Tuple
from dataclasses import dataclass
from datetime import datetime, timedelta
import time

@dataclass
class WindowAnalytics:
    """Results from window function analysis."""
    total_rows: int
    partitions: int
    execution_time_ms: float
    results: pd.DataFrame

class WindowFunctionEngine:
    """Simulates SQL window functions for analytics."""

    def __init__(self, data: pd.DataFrame):
        """Initialize with dataset."""
        self.data = data

    def ranking_functions(
        self, 
        partition_col: str,
        order_col: str,
        ascending: bool = False
    ) -> WindowAnalytics:
        """
        Apply all ranking functions: ROW_NUMBER, RANK, DENSE_RANK.
        """
        start_time = time.time()

        # ROW_NUMBER: Always unique (1, 2, 3, 4, 5...)
        self.data['row_number'] = self.data.groupby(partition_col)[order_col].rank(
            method='first', ascending=ascending
        ).astype(int)

        # RANK: Ties get same rank, next rank skips (1, 1, 3, 4, 4, 6...)
        self.data['rank'] = self.data.groupby(partition_col)[order_col].rank(
            method='min', ascending=ascending
        ).astype(int)

        # DENSE_RANK: Ties get same rank, no gaps (1, 1, 2, 3, 3, 4...)
        self.data['dense_rank'] = self.data.groupby(partition_col)[order_col].rank(
            method='dense', ascending=ascending
        ).astype(int)

        # NTILE: Divide into N buckets (quartiles = 4 buckets)
        self.data['quartile'] = self.data.groupby(partition_col)[order_col].apply(
            lambda x: pd.qcut(x, q=4, labels=['Q1', 'Q2', 'Q3', 'Q4'], duplicates='drop')
        )

        execution_time = (time.time() - start_time) * 1000

        return WindowAnalytics(
            total_rows=len(self.data),
            partitions=self.data[partition_col].nunique(),
            execution_time_ms=execution_time,
            results=self.data
        )

    def lag_lead_analysis(
        self,
        partition_col: str,
        order_col: str,
        value_col: str,
        periods: int = 1
    ) -> pd.DataFrame:
        """
        Apply LAG/LEAD for time-series analysis.

        LAG: Access previous row value
        LEAD: Access next row value
        """
        # Sort within partitions
        sorted_data = self.data.sort_values([partition_col, order_col])

        # LAG: Previous value
        sorted_data['lag_value'] = sorted_data.groupby(partition_col)[value_col].shift(periods)

        # LEAD: Next value
        sorted_data['lead_value'] = sorted_data.groupby(partition_col)[value_col].shift(-periods)

        # Calculate changes
        sorted_data['change_from_prev'] = sorted_data[value_col] - sorted_data['lag_value']
        sorted_data['pct_change'] = (
            (sorted_data[value_col] - sorted_data['lag_value']) / 
            sorted_data['lag_value'] * 100
        ).round(2)

        return sorted_data

    def running_aggregates(
        self,
        order_col: str,
        value_col: str,
        window_size: int = None
    ) -> pd.DataFrame:
        """
        Calculate running totals, averages, and moving windows.

        window_size: If None, cumulative from start. If N, rolling N-period.
        """
        sorted_data = self.data.sort_values(order_col)

        if window_size is None:
            # Cumulative (unbounded)
            sorted_data['running_sum'] = sorted_data[value_col].cumsum()
            sorted_data['running_avg'] = sorted_data[value_col].expanding().mean()
            sorted_data['running_max'] = sorted_data[value_col].cummax()
            sorted_data['running_min'] = sorted_data[value_col].cummin()
        else:
            # Rolling window (bounded)
            sorted_data['rolling_sum'] = sorted_data[value_col].rolling(
                window=window_size, min_periods=1
            ).sum()
            sorted_data['rolling_avg'] = sorted_data[value_col].rolling(
                window=window_size, min_periods=1
            ).mean()
            sorted_data['rolling_std'] = sorted_data[value_col].rolling(
                window=window_size, min_periods=1
            ).std()

        return sorted_data

# ============================================================================
# EXAMPLE 1: NETFLIX - TOP N CONTENT PER GENRE (RANKING)
# ============================================================================
print("="*70)
print("NETFLIX - TOP 5 SHOWS PER GENRE (Window Ranking)")
print("="*70)

np.random.seed(42)

# Netflix content catalog with viewing hours
netflix_content = pd.DataFrame({
    'content_id': range(1, 501),
    'title': [f'Show_{i}' for i in range(1, 501)],
    'genre': np.random.choice(['Drama', 'Comedy', 'Action', 'Sci-Fi', 'Documentary'], 500),
    'total_hours_watched': np.random.randint(100000, 10000000, 500),
    'release_year': np.random.randint(2015, 2025, 500),
    'rating': np.random.uniform(6.0, 9.5, 500).round(2)
})

engine = WindowFunctionEngine(netflix_content)

# Apply ranking within each genre
analytics = engine.ranking_functions(
    partition_col='genre',
    order_col='total_hours_watched',
    ascending=False  # DESC: highest first
)

# Get top 5 per genre
top_5_per_genre = analytics.results[
    analytics.results['row_number'] <= 5
].sort_values(['genre', 'row_number'])

print(f"\n✅ Window Function Analysis:")
print(f"   Total content: {analytics.total_rows:,}")
print(f"   Genres: {analytics.partitions}")
print(f"   Execution time: {analytics.execution_time_ms:.2f}ms")
print(f"\nTop 3 Drama shows:")
drama_top = top_5_per_genre[top_5_per_genre['genre'] == 'Drama'].head(3)
for idx, row in drama_top.iterrows():
    print(f"   #{int(row['row_number'])}: {row['title']} - {row['total_hours_watched']:,} hours")

# SQL Equivalent:
sql_query = """
-- Netflix uses this for "Top 10 in Genre" feature
WITH ranked_content AS (
    SELECT 
        title,
        genre,
        total_hours_watched,
        ROW_NUMBER() OVER (
            PARTITION BY genre 
            ORDER BY total_hours_watched DESC
        ) as rank
    FROM content
)
SELECT * FROM ranked_content WHERE rank <= 5;
"""
print(f"\n📝 SQL Equivalent:\n{sql_query}")

# ============================================================================
# EXAMPLE 2: GOOGLE ADS - LAG/LEAD FOR CAMPAIGN PERFORMANCE
# ============================================================================
print("\n" + "="*70)
print("GOOGLE ADS - Month-over-Month Growth Analysis (LAG/LEAD)")
print("="*70)

# Google Ads monthly performance by campaign
google_campaigns = pd.DataFrame({
    'campaign_id': np.repeat([1, 2, 3], 12),
    'month': pd.date_range('2024-01-01', periods=12, freq='MS').tolist() * 3,
    'impressions': np.random.randint(100000, 1000000, 36),
    'clicks': np.random.randint(1000, 50000, 36),
    'spend': np.random.randint(5000, 100000, 36),
    'conversions': np.random.randint(50, 2000, 36)
})

engine_ga = WindowFunctionEngine(google_campaigns)

# Apply LAG for month-over-month comparison
mom_analysis = engine_ga.lag_lead_analysis(
    partition_col='campaign_id',
    order_col='month',
    value_col='spend',
    periods=1
)

# Also calculate LEAD to see next month
mom_analysis['next_month_spend'] = mom_analysis.groupby('campaign_id')['spend'].shift(-1)

print(f"\n✅ Month-over-Month Analysis (Campaign 1):")
campaign_1 = mom_analysis[mom_analysis['campaign_id'] == 1].head(6)
for idx, row in campaign_1.iterrows():
    month_str = row['month'].strftime('%Y-%m')
    current = row['spend']
    prev = row['lag_value']
    pct = row['pct_change']
    if pd.notna(prev):
        print(f"   {month_str}: ${current:,} (vs ${prev:,.0f} = {pct:+.1f}%)")
    else:
        print(f"   {month_str}: ${current:,} (baseline)")

# SQL Equivalent:
sql_query_lag = """
-- Google Ads dashboard query
SELECT 
    campaign_id,
    month,
    spend as current_spend,
    LAG(spend, 1) OVER (
        PARTITION BY campaign_id 
        ORDER BY month
    ) as prev_month_spend,
    spend - LAG(spend, 1) OVER (
        PARTITION BY campaign_id 
        ORDER BY month
    ) as spend_change,
    ROUND(
        (spend - LAG(spend, 1) OVER (PARTITION BY campaign_id ORDER BY month)) 
        / LAG(spend, 1) OVER (PARTITION BY campaign_id ORDER BY month) * 100,
    2) as pct_change
FROM ad_performance;
"""
print(f"\n📝 SQL Equivalent:\n{sql_query_lag}")

# ============================================================================
# EXAMPLE 3: UBER - MOVING AVERAGES FOR SURGE PRICING
# ============================================================================
print("\n" + "="*70)
print("UBER - 15-Minute Moving Average for Surge Pricing")
print("="*70)

# Uber ride requests per minute
uber_requests = pd.DataFrame({
    'timestamp': pd.date_range('2024-01-01 08:00', periods=60, freq='1min'),
    'requests_per_min': np.random.randint(10, 200, 60),
    'available_drivers': np.random.randint(50, 150, 60)
})

engine_uber = WindowFunctionEngine(uber_requests)

# Calculate 15-minute moving average
surge_analysis = engine_uber.running_aggregates(
    order_col='timestamp',
    value_col='requests_per_min',
    window_size=15  # 15-minute window
)

# Calculate demand/supply ratio
surge_analysis['demand_supply_ratio'] = (
    surge_analysis['rolling_avg'] / surge_analysis['available_drivers']
).round(3)

# Uber applies surge when ratio > 1.5
surge_analysis['surge_multiplier'] = surge_analysis['demand_supply_ratio'].apply(
    lambda x: 1.0 if x < 1.0 else min(x, 3.0)  # Cap at 3.0x
)

print(f"\n✅ Surge Pricing Analysis (sample 10-minute window):")
sample_window = surge_analysis.iloc[15:25]  # Minutes 15-25
for idx, row in sample_window.iterrows():
    time_str = row['timestamp'].strftime('%H:%M')
    requests = row['requests_per_min']
    rolling_avg = row['rolling_avg']
    surge = row['surge_multiplier']
    print(f"   {time_str}: {requests} req/min | 15-min avg: {rolling_avg:.1f} | Surge: {surge:.2f}x")

avg_surge = surge_analysis['surge_multiplier'].mean()
print(f"\n   Average surge multiplier: {avg_surge:.2f}x")
print(f"   Peak surge: {surge_analysis['surge_multiplier'].max():.2f}x")
print(f"   Uber dynamically adjusts pricing every minute using moving averages")

# SQL Equivalent:
sql_query_uber = """
-- Uber surge pricing calculation
SELECT 
    timestamp,
    requests_per_min,
    AVG(requests_per_min) OVER (
        ORDER BY timestamp
        ROWS BETWEEN 14 PRECEDING AND CURRENT ROW
    ) as moving_avg_15min,
    available_drivers,
    AVG(requests_per_min) OVER (
        ORDER BY timestamp
        ROWS BETWEEN 14 PRECEDING AND CURRENT ROW
    ) / available_drivers as demand_supply_ratio
FROM ride_requests
WHERE timestamp >= CURRENT_TIMESTAMP - INTERVAL '1 hour';
"""
print(f"\n📝 SQL Equivalent:\n{sql_query_uber}")

# ============================================================================
# EXAMPLE 4: SPOTIFY - CUMULATIVE STREAMS FOR ROYALTY CALCULATION
# ============================================================================
print("\n" + "="*70)
print("SPOTIFY - Cumulative Streams for Artist Royalties")
print("="*70)

# Artist daily streams
spotify_streams = pd.DataFrame({
    'artist_id': [1001] * 30,
    'date': pd.date_range('2024-01-01', periods=30, freq='D'),
    'daily_streams': np.random.randint(50000, 500000, 30),
    'royalty_rate': 0.004  # $0.004 per stream
})

engine_spotify = WindowFunctionEngine(spotify_streams)

# Calculate cumulative streams and earnings
royalty_analysis = engine_spotify.running_aggregates(
    order_col='date',
    value_col='daily_streams',
    window_size=None  # Cumulative from start
)

royalty_analysis['daily_earnings'] = (
    royalty_analysis['daily_streams'] * royalty_analysis['royalty_rate']
)
royalty_analysis['cumulative_earnings'] = (
    royalty_analysis['running_sum'] * royalty_analysis['royalty_rate']
)

print(f"\n✅ Artist Royalty Tracking (First 10 days):")
first_10_days = royalty_analysis.head(10)
for idx, row in first_10_days.iterrows():
    date_str = row['date'].strftime('%Y-%m-%d')
    daily = row['daily_streams']
    cumulative = row['running_sum']
    earnings = row['cumulative_earnings']
    print(f"   {date_str}: {daily:,} streams | Total: {cumulative:,.0f} | Earned: ${earnings:,.2f}")

total_earnings = royalty_analysis['cumulative_earnings'].iloc[-1]
total_streams = royalty_analysis['running_sum'].iloc[-1]
print(f"\n   30-day totals:")
print(f"   Total streams: {total_streams:,.0f}")
print(f"   Total royalties: ${total_earnings:,.2f}")
print(f"   Spotify pays artists monthly using cumulative sum calculations")

print("\n" + "="*70)
print("Key Takeaway: Window functions preserve row-level detail!")
print("="*70)

-- ====================================================================
-- PATTERN 1: Top N per group (Netflix "Top 10 in Genre")
-- ====================================================================
WITH ranked AS (
    SELECT 
        genre,
        title,
        watch_hours,
        ROW_NUMBER() OVER (
            PARTITION BY genre 
            ORDER BY watch_hours DESC
        ) as rank_in_genre
    FROM content_performance
    WHERE date >= CURRENT_DATE - INTERVAL '7 days'
)
SELECT * FROM ranked WHERE rank_in_genre <= 10;

-- Netflix updates this hourly for 190M users

-- ====================================================================
-- PATTERN 2: Gap analysis (Find missing dates in sequence)
-- ====================================================================
WITH sequences AS (
    SELECT 
        date,
        LAG(date) OVER (ORDER BY date) as prev_date,
        date - LAG(date) OVER (ORDER BY date) as day_gap
    FROM daily_metrics
)
SELECT * FROM sequences WHERE day_gap > 1;

-- Airbnb uses this to detect data pipeline failures

-- ====================================================================
-- PATTERN 3: Running vs. Comparative metrics
-- ====================================================================
SELECT 
    month,
    revenue,

    -- Running total (cumulative)
    SUM(revenue) OVER (
        ORDER BY month
        ROWS BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW
    ) as cumulative_revenue,

    -- 3-month moving average
    AVG(revenue) OVER (
        ORDER BY month
        ROWS BETWEEN 2 PRECEDING AND CURRENT ROW
    ) as ma_3month,

    -- YoY comparison
    LAG(revenue, 12) OVER (ORDER BY month) as same_month_last_year,

    -- Percentage of total
    revenue / SUM(revenue) OVER () * 100 as pct_of_total,

    -- Rank this month vs all months
    RANK() OVER (ORDER BY revenue DESC) as revenue_rank

FROM monthly_sales;

-- Stripe uses this pattern for merchant analytics dashboards

-- ====================================================================
-- PATTERN 4: Identify consecutive events
-- ====================================================================
WITH event_groups AS (
    SELECT 
        user_id,
        event_date,
        event_date - ROW_NUMBER() OVER (
            PARTITION BY user_id 
            ORDER BY event_date
        )::int as group_id
    FROM user_logins
)
SELECT 
    user_id,
    MIN(event_date) as streak_start,
    MAX(event_date) as streak_end,
    COUNT(*) as consecutive_days
FROM event_groups
GROUP BY user_id, group_id
HAVING COUNT(*) >= 7;  -- 7+ day streaks

-- Duolingo uses this for "streak" gamification feature

-- ====================================================================
-- PATTERN 5: NTILE for percentile bucketing
-- ====================================================================
SELECT 
    customer_id,
    total_spend,
    NTILE(100) OVER (ORDER BY total_spend) as percentile,
    NTILE(4) OVER (ORDER BY total_spend) as quartile,
    CASE 
        WHEN NTILE(10) OVER (ORDER BY total_spend) = 10 THEN 'Top 10%'
        WHEN NTILE(10) OVER (ORDER BY total_spend) >= 8 THEN 'Top 20%'
        ELSE 'Other'
    END as segment
FROM customer_lifetime_value;

-- Amazon targets top 20% customers with Prime benefits

┌──────────────────────────────────────────────────────────────────────┐
│              Duplicate Detection: METHOD COMPARISON                    │
├──────────────────────────────────────────────────────────────────────┤
│                                                                      │
│  METHOD 1: GROUP BY + HAVING (Simple, COUNT only)                    │
│  ┌────────────────────────────────────────────────────────────────┐ │
│  │  SELECT email, COUNT(*) FROM users                             │ │
│  │  GROUP BY email HAVING COUNT(*) > 1                            │ │
│  │                                                                │ │
│  │  Result: email | count                                         │ │
│  │          [email protected] | 3                                     │ │
│  │                                                                │ │
│  │  ✅ Simple, fast                                                │ │
│  │  ❌ Can't see individual rows                                   │ │
│  └────────────────────────────────────────────────────────────────┘ │
│                                                                      │
│  METHOD 2: ROW_NUMBER (Modern, powerful)                             │
│  ┌────────────────────────────────────────────────────────────────┐ │
│  │  WITH ranked AS (                                              │ │
│  │    SELECT *, ROW_NUMBER() OVER (                               │ │
│  │      PARTITION BY email ORDER BY created_at                    │ │
│  │    ) as rn FROM users                                          │ │
│  │  ) SELECT * FROM ranked WHERE rn > 1                           │ │
│  │                                                                │ │
│  │  ✅ See all duplicate rows with details                         │ │
│  │  ✅ Control which to keep (ORDER BY)                            │ │
│  │  ✅ Works with DELETE                                           │ │
│  └────────────────────────────────────────────────────────────────┘ │
│                                                                      │
└──────────────────────────────────────────────────────────────────────┘

import pandas as pd
import numpy as np
from typing import List, Dict, Tuple
from dataclasses import dataclass
from collections import defaultdict

@dataclass
class DuplicateMetrics:
    """Track duplicate detection results."""
    total_rows: int
    unique_rows: int
    duplicate_rows: int
    duplicate_groups: int
    duplication_rate: float
    data_quality_score: float

class DeduplicationEngine:
    """Simulate SQL duplicate detection and removal strategies."""

    def find_duplicates_group_by(self, df: pd.DataFrame, 
                                 key_cols: List[str]) -> Tuple[pd.DataFrame, DuplicateMetrics]:
        """
        GROUP BY + HAVING approach.

        SQL equivalent:
        SELECT key_cols, COUNT(*) as count
        FROM table
        GROUP BY key_cols
        HAVING COUNT(*) > 1;
        """
        # Group and count
        grouped = df.groupby(key_cols).size().reset_index(name='count')
        duplicates = grouped[grouped['count'] > 1]

        total = len(df)
        unique = len(grouped)
        dup_groups = len(duplicates)
        dup_rows = duplicates['count'].sum() - dup_groups  # Extras only

        metrics = DuplicateMetrics(
            total_rows=total,
            unique_rows=unique,
            duplicate_rows=dup_rows,
            duplicate_groups=dup_groups,
            duplication_rate=(dup_rows / total * 100) if total > 0 else 0,
            data_quality_score=((total - dup_rows) / total * 100) if total > 0 else 100
        )

        return duplicates, metrics

    def find_duplicates_row_number(self, df: pd.DataFrame,
                                   partition_cols: List[str],
                                   order_col: str) -> Tuple[pd.DataFrame, DuplicateMetrics]:
        """
        ROW_NUMBER approach.

        SQL equivalent:
        WITH ranked AS (
            SELECT *, ROW_NUMBER() OVER (
                PARTITION BY partition_cols
                ORDER BY order_col
            ) as rn
            FROM table
        )
        SELECT * FROM ranked WHERE rn > 1;
        """
        # Add row number within partition
        df_ranked = df.copy()
        df_ranked['rn'] = df_ranked.groupby(partition_cols)[order_col].rank(method='first')

        # Get duplicates (rn > 1)
        duplicates = df_ranked[df_ranked['rn'] > 1].copy()

        total = len(df)
        dup_rows = len(duplicates)
        unique = total - dup_rows
        dup_groups = df.duplicated(subset=partition_cols).sum()

        metrics = DuplicateMetrics(
            total_rows=total,
            unique_rows=unique,
            duplicate_rows=dup_rows,
            duplicate_groups=dup_groups,
            duplication_rate=(dup_rows / total * 100) if total > 0 else 0,
            data_quality_score=((total - dup_rows) / total * 100) if total > 0 else 100
        )

        return duplicates, metrics

    def remove_duplicates(self, df: pd.DataFrame,
                        subset_cols: List[str],
                        keep: str = 'first') -> Tuple[pd.DataFrame, int]:
        """
        Remove duplicates keeping specified occurrence.

        SQL equivalent:
        WITH ranked AS (
            SELECT *, ROW_NUMBER() OVER (
                PARTITION BY subset_cols ORDER BY created_at
            ) as rn
            FROM table
        )
        DELETE FROM table WHERE id IN (
            SELECT id FROM ranked WHERE rn > 1
        );
        """
        before_count = len(df)
        deduplicated = df.drop_duplicates(subset=subset_cols, keep=keep)
        after_count = len(deduplicated)
        removed = before_count - after_count

        return deduplicated, removed

# ========================================================================
# EXAMPLE 1: LINKEDIN - USER PROFILE DEDUPLICATION
# ========================================================================
print("="*70)
print("LINKEDIN - Detecting Duplicate User Profiles (12% Duplicates!)")
print("="*70)

# LinkedIn user data with duplicates
linkedin_users = pd.DataFrame({
    'user_id': range(1, 26),
    'email': [
        '[email protected]', '[email protected]', '[email protected]',
        '[email protected]', '[email protected]', '[email protected]',
        '[email protected]', '[email protected]', '[email protected]',
        '[email protected]', '[email protected]', '[email protected]',
        '[email protected]', '[email protected]', '[email protected]',
        '[email protected]', '[email protected]', '[email protected]',
        '[email protected]', '[email protected]', '[email protected]',
        '[email protected]', '[email protected]', '[email protected]', '[email protected]'
    ],
    'name': [
        'John Doe', 'John Doe', 'Jane Smith',
        'Bob Johnson', 'Alice Brown', 'Alice Brown',
        'Charlie Davis', 'Dave Wilson', 'Eve Martinez',
        'Frank Anderson', 'Frank Anderson', 'Frank Anderson',
        'Grace Thomas', 'Henry Taylor', 'Ivy Moore',
        'Jack Jackson', 'Jack Jackson', 'Kate White',
        'Leo Harris', 'Mary Martin', 'Nancy Thompson',
        'Oscar Garcia', 'Paul Rodriguez', 'Paul Rodriguez', 'Quinn Lee'
    ],
    'created_at': pd.date_range('2024-01-01', periods=25, freq='D')
})

engine = DeduplicationEngine()

print(f"\n📊 Dataset: {len(linkedin_users)} user profiles")

# METHOD 1: GROUP BY (count only)
dup_summary, metrics_group = engine.find_duplicates_group_by(
    linkedin_users, ['email']
)

print(f"\n🔍 METHOD 1: GROUP BY + HAVING")
print(f"   Duplicate email addresses:")
for _, row in dup_summary.iterrows():
    print(f"     {row['email']}: {row['count']} profiles")

print(f"\n   📊 Data Quality Metrics:")
print(f"     Total profiles:        {metrics_group.total_rows}")
print(f"     Unique emails:         {metrics_group.unique_rows}")
print(f"     Duplicate groups:      {metrics_group.duplicate_groups}")
print(f"     Duplication rate:      {metrics_group.duplication_rate:.1f}%")
print(f"     Data quality score:    {metrics_group.data_quality_score:.1f}%")

# METHOD 2: ROW_NUMBER (see all duplicates)
dup_details, metrics_rn = engine.find_duplicates_row_number(
    linkedin_users, ['email'], 'created_at'
)

print(f"\n🔍 METHOD 2: ROW_NUMBER (All duplicate rows)")
print(f"   Found {len(dup_details)} duplicate profiles to remove:")
for _, row in dup_details.head(5).iterrows():
    print(f"     User {row['user_id']}: {row['name']} ({row['email']}) - Rank {int(row['rn'])}")

# DEDUPLICATION: Remove keeping oldest
cleaned, removed = engine.remove_duplicates(
    linkedin_users, ['email'], keep='first'
)

print(f"\n✅ DEDUPLICATION Complete:")
print(f"   Removed:  {removed} duplicate profiles")
print(f"   Kept:     {len(cleaned)} unique profiles")
print(f"   Strategy: Kept OLDEST profile per email (first created_at)")

sql_linkedin = """
-- LinkedIn duplicate detection and removal

-- Step 1: Find duplicate email counts
SELECT email, COUNT(*) as profile_count
FROM user_profiles
GROUP BY email
HAVING COUNT(*) > 1;

-- Step 2: See all duplicate profiles with details
WITH ranked_profiles AS (
    SELECT 
        user_id,
        email,
        name,
        created_at,
        ROW_NUMBER() OVER (
            PARTITION BY email 
            ORDER BY created_at  -- Keep oldest
        ) as rn
    FROM user_profiles
)
SELECT * FROM ranked_profiles WHERE rn > 1;

-- Step 3: Remove duplicates (keep oldest)
WITH ranked_profiles AS (
    SELECT 
        user_id,
        ROW_NUMBER() OVER (
            PARTITION BY email 
            ORDER BY created_at
        ) as rn
    FROM user_profiles
)
DELETE FROM user_profiles
WHERE user_id IN (
    SELECT user_id FROM ranked_profiles WHERE rn > 1
);

-- LinkedIn cleaned 12% duplicates (1.2M profiles) in 2019
"""
print(f"\n📝 SQL Equivalent:\n{sql_linkedin}")

# ========================================================================
# EXAMPLE 2: AMAZON - MULTI-COLUMN DUPLICATES
# ========================================================================
print("\n" + "="*70)
print("AMAZON - Product Listing Duplicates (Multi-Column Detection)")
print("="*70)

# Amazon product listings with subtle duplicates
amazon_products = pd.DataFrame({
    'listing_id': range(1, 11),
    'product_name': [
        'iPhone 15', 'iPhone 15', 'iPhone 15 Pro',
        'MacBook Pro', 'MacBook Pro', 'MacBook Air',
        'AirPods Pro', 'AirPods Pro', 'AirPods Max',
        'iPad Pro'
    ],
    'seller_id': [101, 101, 102, 103, 103, 104, 105, 106, 107, 108],
    'price': [999, 999, 1199, 2499, 2499, 1299, 249, 249, 549, 799],
    'condition': ['New', 'New', 'New', 'New', 'New', 'New', 'New', 'New', 'New', 'New']
})

print(f"\n📊 Amazon Listings: {len(amazon_products)} products")

# Multi-column duplicate detection
dup_products, metrics_prod = engine.find_duplicates_group_by(
    amazon_products, ['product_name', 'seller_id', 'price']
)

print(f"\n🔍 Multi-Column Duplicate Detection:")
print(f"   Checking: product_name + seller_id + price")
print(f"\n   Duplicate listings found:")
for _, row in dup_products.iterrows():
    print(f"     {row['product_name']} by Seller {row['seller_id']} at ${row['price']}: {row['count']}x")

print(f"\n   Impact: {metrics_prod.duplicate_rows} duplicate listings")
print(f"   Amazon removes ~500K duplicate listings daily")

sql_amazon = """
-- Amazon multi-column duplicate detection
SELECT 
    product_name,
    seller_id,
    price,
    COUNT(*) as duplicate_count
FROM product_listings
GROUP BY product_name, seller_id, price
HAVING COUNT(*) > 1;

-- Remove with ROW_NUMBER (keep newest)
WITH ranked AS (
    SELECT 
        listing_id,
        ROW_NUMBER() OVER (
            PARTITION BY product_name, seller_id, price
            ORDER BY created_at DESC  -- Keep NEWEST
        ) as rn
    FROM product_listings
)
DELETE FROM product_listings
WHERE listing_id IN (
    SELECT listing_id FROM ranked WHERE rn > 1
);
"""
print(f"\n📝 SQL Equivalent:\n{sql_amazon}")

print("\n" + "="*70)
print("Key Takeaway: ROW_NUMBER gives full control over deduplication!")
print("="*70)