Distributed SQL Query Engines

Trino Query Execution:

┌────────────────────────────────────────────────────┐
│                  COORDINATOR                       │
│                                                    │
│  1. SQL Parser → Abstract Syntax Tree (AST)        │
│  2. Analyzer → Validate & resolve references       │
│  3. Planner → Generate optimized execution plan    │
│  4. Scheduler → Assign tasks to workers            │
│  5. Coordinator → Aggregate final results          │
└───────────────┬────────────────────────────────────┘
                │
        ┌───────┴────────┬──────────────┐
        ▼                ▼              ▼
    ┌───────┐        ┌───────┐      ┌───────┐
    │Worker1│        │Worker2│      │WorkerN│
    │       │        │       │      │       │
    │Split 1│        │Split 2│      │Split N│
    │(1GB)  │        │(1GB)  │      │(1GB)  │
    └───┬───┘        └───┬───┘      └───┬───┘
        │                │              │
        ▼                ▼              ▼
    ┌────────────────────────────────────┐
    │         Data Sources               │
    │  Hive, S3, PostgreSQL, MySQL, etc. │
    └────────────────────────────────────┘

KEY CONCEPTS:
• Splits: Data chunks (typically 1GB) processed in parallel
• Stages: Pipeline of operations (scan → filter → aggregate)
• Pipelining: Data flows through operators without disk writes
• Cost-based optimization: Chooses optimal join order

from trino.dbapi import connect
import pandas as pd

# Connect to Trino
conn = connect(
    host='trino.example.com',
    port=8080,
    user='analyst',
    catalog='hive',  # Which data source
    schema='default'  # Which database/schema
)

cursor = conn.cursor()

# ============ SIMPLE QUERY ============
query = """
    SELECT
        customer_id,
        COUNT(*) as order_count,
        SUM(amount) as total_spent
    FROM sales
    WHERE sale_date >= DATE '2024-01-01'
    GROUP BY customer_id
    ORDER BY total_spent DESC
    LIMIT 10
"""

cursor.execute(query)
results = cursor.fetchall()

# Convert to DataFrame
df = pd.DataFrame(results, columns=[desc[0] for desc in cursor.description])
print(df)

"""
+-------------+-------------+-------------+
| customer_id | order_count | total_spent |
+-------------+-------------+-------------+
|        1234 |          45 |     12500.0 |
|        5678 |          38 |     11200.0 |
|        9012 |          42 |     10800.0 |
|         ... |             |             |
+-------------+-------------+-------------+

Query completed in 3.2 seconds
Scanned 1.2 TB, processed 450M rows
"""

# ============ FEDERATED QUERY (multiple sources) ============
# Join S3 data with PostgreSQL customer info
federated_query = """
    SELECT
        s.sale_id,
        s.amount,
        c.name,
        c.email
    FROM hive.default.sales s              -- Data on S3
    JOIN postgresql.public.customers c     -- Data in PostgreSQL
        ON s.customer_id = c.customer_id
    WHERE s.sale_date >= CURRENT_DATE - INTERVAL '7' DAY
"""

cursor.execute(federated_query)
results = cursor.fetchall()

"""
Result: Joins 100GB S3 data with 10GB PostgreSQL data
Completed in 12 seconds

Execution plan:
1. Scan S3 sales (filtered by date)
2. Scan PostgreSQL customers
3. Hash join on customer_id
4. Return results
"""

# ============ COMPLEX ANALYTICS ============
# Window functions, CTEs, aggregations
complex_query = """
    WITH daily_sales AS (
        SELECT
            DATE(sale_date) as day,
            SUM(amount) as daily_total
        FROM sales
        WHERE sale_date >= DATE '2024-01-01'
        GROUP BY DATE(sale_date)
    ),
    moving_avg AS (
        SELECT
            day,
            daily_total,
            AVG(daily_total) OVER (
                ORDER BY day
                ROWS BETWEEN 6 PRECEDING AND CURRENT ROW
            ) as seven_day_avg
        FROM daily_sales
    )
    SELECT *
    FROM moving_avg
    WHERE daily_total > seven_day_avg * 1.2  -- 20% above average
    ORDER BY day DESC
"""

cursor.execute(complex_query)
anomalies = cursor.fetchall()

print(f"Found {len(anomalies)} anomalous days (sales 20% above 7-day average)")

conn.close()

# ============ PARTITION PRUNING ============
# BAD: Scans entire table (10 TB)
SELECT * FROM sales WHERE customer_id = 1234

# GOOD: Filters by partition column first (10 GB)
SELECT * FROM sales
WHERE year = 2024
  AND month = 1
  AND customer_id = 1234

# Result: 1000x less data scanned


# ============ PREDICATE PUSHDOWN ============
# Trino pushes filters to data source (avoid reading unnecessary data)
SELECT * FROM postgresql.public.users
WHERE created_at >= '2024-01-01'

# Execution:
# 1. Trino sends filter to PostgreSQL
# 2. PostgreSQL applies filter before returning data
# 3. Only filtered rows transferred to Trino


# ============ JOIN ORDERING ============
# Trino's cost-based optimizer reorders joins

# You write:
SELECT *
FROM large_table   -- 1 TB
JOIN small_table   -- 1 GB
  ON large_table.id = small_table.id

# Trino executes:
# 1. Broadcast small_table to all workers (cheap)
# 2. Hash join with large_table (no shuffle)
# Result: Much faster than shuffling 1 TB table


# ============ COLUMN PRUNING ============
# BAD: Reads all columns from Parquet
SELECT customer_id FROM sales

# GOOD: Parquet is columnar, Trino only reads customer_id column
# Result: Read 100 MB instead of 10 GB


# ============ EXPLAIN PLAN ============
# View query execution plan
EXPLAIN SELECT * FROM sales WHERE year = 2024

"""
Fragment 0 [COORDINATOR]
    Output: customer_id, amount, sale_date
    └─ Exchange (gather)
        └─ Fragment 1 [SOURCE]
            └─ TableScan[hive.default.sales]
                Estimates: {rows: 1000000 (100MB), cpu: 100M, memory: 0B, network: 0B}
                Partition filters: (year = 2024)
                Columns: customer_id, amount, sale_date
"""

# Shows:
# - How data flows through stages
# - Estimated rows/memory/CPU
# - Which filters applied
# - Join strategies used

from impala.dbapi import connect
import pandas as pd

# Connect to Impala
conn = connect(
    host='impala-coordinator.example.com',
    port=21050,
    database='default'
)

cursor = conn.cursor()

# ============ REFRESH METADATA ============
# Impala caches metadata; refresh after new data arrives
cursor.execute("INVALIDATE METADATA sales")
cursor.execute("REFRESH sales")

# ============ QUERY EXECUTION ============
query = """
    SELECT
        product_category,
        COUNT(DISTINCT customer_id) as unique_customers,
        SUM(amount) as revenue
    FROM sales
    WHERE year = 2024 AND month = 1
    GROUP BY product_category
    ORDER BY revenue DESC
"""

cursor.execute(query)
results = cursor.fetchall()

df = pd.DataFrame(results, columns=[desc[0] for desc in cursor.description])
print(df)

"""
+------------------+------------------+------------+
| product_category | unique_customers |    revenue |
+------------------+------------------+------------+
|      Electronics |             2500 | 1250000.00 |
|         Clothing |             1800 |  850000.00 |
|             Home |             1200 |  620000.00 |
+------------------+------------------+------------+

Query completed in 2.1 seconds
Scanned 50 GB across 200 partitions
"""

# ============ QUERY PROFILE ============
# Get detailed execution stats
cursor.execute("PROFILE")
profile = cursor.fetchall()

# Profile shows:
# - Time spent in each operator
# - Rows processed per node
# - Memory usage
# - Network I/O

conn.close()

Athena Architecture:

┌─────────────────────────────────────┐
│      User (SQL Query via UI/API)    │
└─────────────────┬───────────────────┘
                  │
┌─────────────────▼───────────────────┐
│         AWS Athena Service          │
│  • Serverless (no infra to manage)  │
│  • Auto-scaling workers             │
│  • Based on Trino/Presto            │
└─────────────────┬───────────────────┘
                  │
        ┌─────────┴─────────┐
        ▼                   ▼
┌──────────────┐    ┌──────────────┐
│  S3 Bucket   │    │ Glue Catalog │
│  (Data Lake) │    │  (Metadata)  │
└──────────────┘    └──────────────┘

PRICING MODEL:
• $5 per TB scanned (us-east-1)
• No charge for DDL (CREATE TABLE, etc.)
• No charge for failed queries
• No infrastructure costs

COST OPTIMIZATION:
• Partition data: Reduce scanned data by 100x
• Use Parquet/ORC: 5-10x less scanned vs JSON/CSV
• Compress data: Further reduce scan size
• SELECT only needed columns: Columnar formats read only specified columns

Example Cost:
Query 1 TB JSON → $5
Query 100 GB Parquet (partitioned) → $0.50 (10x savings)

import boto3
from pyathena import connect
import pandas as pd

# ============ METHOD 1: PyAthena (DBAPI) ============
conn = connect(
    s3_staging_dir='s3://my-bucket/athena-results/',
    region_name='us-east-1'
)

cursor = conn.cursor()

query = """
    SELECT
        product_id,
        SUM(quantity) as total_sold,
        SUM(amount) as total_revenue
    FROM sales
    WHERE year = 2024
      AND month = 1
    GROUP BY product_id
    HAVING total_revenue > 10000
    ORDER BY total_revenue DESC
    LIMIT 100
"""

# Execute query (asynchronous by default)
cursor.execute(query)

# Fetch results
results = cursor.fetchall()
df = pd.DataFrame(results, columns=[desc[0] for desc in cursor.description])

print(df)
"""
+------------+------------+---------------+
| product_id | total_sold | total_revenue |
+------------+------------+---------------+
|       1001 |        450 |      125000.0 |
|       2003 |        380 |      110000.0 |
|       5012 |        290 |       95000.0 |
+------------+------------+---------------+

Query statistics:
• Data scanned: 12.5 GB
• Execution time: 4.2 seconds
• Cost: ~$0.06
"""

# ============ METHOD 2: Pandas Integration ============
# Query directly to DataFrame
df = pd.read_sql(query, conn)

# ============ METHOD 3: Boto3 (AWS SDK) ============
# More control over execution
athena_client = boto3.client('athena', region_name='us-east-1')

# Start query execution
response = athena_client.start_query_execution(
    QueryString=query,
    QueryExecutionContext={'Database': 'my_database'},
    ResultConfiguration={
        'OutputLocation': 's3://my-bucket/athena-results/'
    }
)

query_execution_id = response['QueryExecutionId']

# Wait for query to complete
import time
while True:
    result = athena_client.get_query_execution(QueryExecutionId=query_execution_id)
    state = result['QueryExecution']['Status']['State']

    if state in ['SUCCEEDED', 'FAILED', 'CANCELLED']:
        break

    time.sleep(1)

# Get execution statistics
stats = result['QueryExecution']['Statistics']
print(f"Data scanned: {stats['DataScannedInBytes'] / 1e9:.2f} GB")
print(f"Execution time: {stats['TotalExecutionTimeInMillis'] / 1000:.2f} seconds")
print(f"Cost: ${stats['DataScannedInBytes'] / 1e12 * 5:.4f}")

# Read results from S3 (parse the S3 URL)
result_location = result['QueryExecution']['ResultConfiguration']['OutputLocation']
bucket = result_location.split('/')[2]
key = '/'.join(result_location.split('/')[3:])
obj = boto3.client('s3').get_object(Bucket=bucket, Key=key)
df = pd.read_csv(obj['Body'])

# ============ PARTITIONING (Critical for cost optimization) ============
# Create partitioned table
CREATE EXTERNAL TABLE sales_partitioned (
    order_id STRING,
    customer_id STRING,
    amount DOUBLE
)
PARTITIONED BY (year INT, month INT, day INT)
STORED AS PARQUET
LOCATION 's3://my-bucket/sales/'

# Add partitions (after loading data)
MSCK REPAIR TABLE sales_partitioned

# Query with partition filter (scans only relevant files)
SELECT * FROM sales_partitioned
WHERE year = 2024 AND month = 1 AND day = 15
-- Scans: 1 day = 100 GB
-- Without partition filter: 100 TB (1000x more!)


# ============ CTAS (CREATE TABLE AS SELECT) ============
# Convert JSON to optimized Parquet
CREATE TABLE sales_optimized
WITH (
    format = 'PARQUET',
    parquet_compression = 'SNAPPY',
    partitioned_by = ARRAY['year', 'month'],
    bucketed_by = ARRAY['customer_id'],
    bucket_count = 100
)
AS
SELECT *
FROM sales_json
WHERE year >= 2024

# Result: 10x smaller, 10x faster queries, 100x cost savings


# ============ FEDERATED QUERIES ============
# Query across S3, RDS, DynamoDB in single query
SELECT
    s.order_id,
    s.amount,
    c.name,
    d.loyalty_points
FROM "hive".default.sales s                -- S3 data
JOIN "postgres".public.customers c         -- RDS PostgreSQL
    ON s.customer_id = c.customer_id
LEFT JOIN "dynamodb".default.loyalty d     -- DynamoDB
    ON s.customer_id = d.customer_id
WHERE s.year = 2024


# ============ QUERY RESULT REUSE ============
# Athena caches results for 7 days
# Re-running identical query = instant results, $0 cost


# ============ WORKGROUPS (Cost control & governance) ============
# Create workgroup with limits
aws athena create-work-group --name analytics-team \
    --configuration "EnforceWorkGroupConfiguration=true,\
                     BytesScannedCutoffPerQuery=1000000000000"  # 1 TB limit

# Queries exceeding limit are automatically cancelled