Athena CTAS Patterns for MCP Analytics Tools

Partitioned CTAS for MCP daily aggregation tables

A daily materialized summary table that MCP reporting tools query instead of the raw event table:

import boto3
import time
from datetime import date, timedelta

athena = boto3.client("athena", region_name="us-east-1")

def run_ctas(sql: str, workgroup: str = "mcp-etl-jobs") -> str:
    """Run CTAS statement; return query execution ID."""
    resp = athena.start_query_execution(
        QueryString=sql,
        WorkGroup=workgroup,
    )
    qid = resp["QueryExecutionId"]
    while True:
        exec_resp = athena.get_query_execution(QueryExecutionId=qid)
        state = exec_resp["QueryExecution"]["Status"]["State"]
        if state == "SUCCEEDED":
            stats = exec_resp["QueryExecution"]["Statistics"]
            scanned_gb = stats.get("DataScannedInBytes", 0) / (1024**3)
            print(f"CTAS completed: {scanned_gb:.2f} GB scanned")
            return qid
        elif state in ("FAILED", "CANCELLED"):
            raise RuntimeError(
                f"CTAS failed: {exec_resp['QueryExecution']['Status'].get('StateChangeReason')}"
            )
        time.sleep(5)   # CTAS jobs are longer — poll less frequently

def materialize_daily_summary(target_date: date) -> None:
    """Materialize MCP event summary for a given date partition."""
    dt_str = target_date.isoformat()   # '2026-10-01'
    s3_prefix = f"s3://my-datalake/materialized/mcp_daily_summary/dt={dt_str}/"

    # Step 1: Drop existing partition data at this location
    # (CTAS fails if external_location already has objects — must clear first)
    s3 = boto3.client("s3")
    paginator = s3.get_paginator("list_objects_v2")
    bucket, prefix = "my-datalake", f"materialized/mcp_daily_summary/dt={dt_str}/"
    for page in paginator.paginate(Bucket=bucket, Prefix=prefix):
        objects = page.get("Contents", [])
        if objects:
            s3.delete_objects(
                Bucket=bucket,
                Delete={"Objects": [{"Key": obj["Key"]} for obj in objects]},
            )

    # Step 2: Drop old Glue partition so the CTAS can recreate it
    glue = boto3.client("glue", region_name="us-east-1")
    try:
        glue.delete_partition(
            DatabaseName="analytics",
            TableName="mcp_daily_summary",
            PartitionValues=[dt_str],
        )
    except glue.exceptions.EntityNotFoundException:
        pass   # partition didn't exist yet — first run for this date

    # Step 3: CTAS into the partitioned external location
    run_ctas(f"""
    CREATE TABLE analytics.mcp_daily_summary_temp_{dt_str.replace('-', '')}
    WITH (
        format                = 'PARQUET',
        parquet_compression   = 'SNAPPY',
        external_location     = '{s3_prefix}',
        partitioned_by        = ARRAY['dt']
    )
    AS
    SELECT
        tool_name,
        status,
        COUNT(*)                                             AS call_count,
        COUNT(*) FILTER (WHERE status = 'success')          AS success_count,
        COUNT(*) FILTER (WHERE status != 'success')         AS error_count,
        AVG(latency_ms)                                      AS avg_latency_ms,
        APPROX_PERCENTILE(latency_ms, 0.95)                  AS p95_latency_ms,
        APPROX_PERCENTILE(latency_ms, 0.99)                  AS p99_latency_ms,
        '{dt_str}'                                           AS dt
    FROM analytics.mcp_events
    WHERE event_ts >= TIMESTAMP '{dt_str} 00:00:00'
      AND event_ts <  TIMESTAMP '{dt_str} 00:00:00' + INTERVAL '1' DAY
    GROUP BY tool_name, status
    """)

    # Step 4: Rename temp CTAS table to the canonical name (or just use canonical name directly)
    # Simpler: use INSERT OVERWRITE INTO if using an Iceberg table as the target,
    # or register the new partition manually via MSCK or Glue AddPartition
    glue.create_partition(
        DatabaseName="analytics",
        TableName="mcp_daily_summary",
        PartitionInput={
            "Values": [dt_str],
            "StorageDescriptor": {
                "Location": s3_prefix,
                "InputFormat": "org.apache.hadoop.hive.ql.io.parquet.MapredParquetInputFormat",
                "OutputFormat": "org.apache.hadoop.hive.ql.io.parquet.MapredParquetOutputFormat",
                "SerdeInfo": {"SerializationLibrary": "org.apache.hadoop.hive.ql.io.parquet.serde.ParquetHiveSerDe"},
            },
        },
    )

Format conversion CTAS for upstream CSV data

Raw event data arrives as gzipped CSV from upstream producers. Convert to Parquet partitioned by date to cut downstream scan costs by 10–50×:

# Convert CSV landing zone to partitioned Parquet
# The CSV table is registered as a Hive external table pointing at s3://raw-events/
run_ctas("""
CREATE TABLE analytics.mcp_events_parquet
WITH (
    format                 = 'PARQUET',
    parquet_compression    = 'SNAPPY',
    external_location      = 's3://my-datalake/events-parquet/',
    partitioned_by         = ARRAY['event_date'],
    -- bucketing reduces join shuffle for large JOINs on user_id
    bucketed_by            = ARRAY['user_id'],
    bucket_count           = 64
)
AS
SELECT
    event_id,
    session_id,
    tool_name,
    user_id,
    CAST(event_ts AS TIMESTAMP)         AS event_ts,
    tool_payload,
    status,
    CAST(latency_ms AS BIGINT)          AS latency_ms,
    CAST(event_ts AS DATE)              AS event_date    -- partition column must be last in SELECT
FROM analytics.mcp_events_csv_raw
WHERE event_ts >= TIMESTAMP '2026-01-01 00:00:00'
""")

# After format conversion, MCP tool queries against mcp_events_parquet:
# - Scan 10-50x less data than the raw CSV (columnar projection + compression)
# - Partition pruning via event_date filter
# - Bucket pruning for JOINs on user_id if the other side is also bucketed 64 ways

CTAS bucketing is only effective when both sides of a JOIN are bucketed on the same column with the same bucket count. For MCP analytics tools that frequently join event data with a user dimension table, bucket both tables on user_id with the same bucket_count to enable bucket-side joins — eliminating the shuffle phase and reducing query time for large joins from minutes to seconds.

UNLOAD for MCP export tools without Glue registration

When an MCP tool needs to write query results to S3 for downstream consumption (another service, a customer data export, a Redshift COPY), use UNLOAD instead of CTAS. UNLOAD writes files without registering a table in Glue:

# UNLOAD: write results to S3 without creating a Glue table
# Use for: customer exports, Redshift COPY sources, downstream ML training sets
run_ctas("""
UNLOAD (
    SELECT
        user_id,
        tool_name,
        event_ts,
        tool_payload,
        status
    FROM analytics.mcp_events
    WHERE event_ts >= TIMESTAMP '2026-09-01 00:00:00'
      AND user_id = 'tenant-acme'
)
TO 's3://customer-exports/acme/mcp-events-2026-09/'
WITH (
    format               = 'PARQUET',
    compression          = 'SNAPPY',
    partitioned_by       = ARRAY['status']
)
""")

# UNLOAD vs CTAS decision:
# Use CTAS when:
#   - Result needs to be queryable via Athena going forward
#   - You want automatic Glue catalog registration
#   - Building materialized views for repeated reads
#
# Use UNLOAD when:
#   - Writing a one-time export for a customer or downstream service
#   - You don't want to litter the Glue catalog with transient tables
#   - The result location is a single customer's isolated S3 prefix
#   - Writing training data for a one-off ML job

# Clean up CTAS temp tables after use (they accumulate in Glue catalog)
def drop_ctas_table(database: str, table: str) -> None:
    """Drop a CTAS-created external table from Glue (data files remain on S3)."""
    glue = boto3.client("glue", region_name="us-east-1")
    try:
        glue.delete_table(DatabaseName=database, Name=table)
    except glue.exceptions.EntityNotFoundException:
        pass

Incremental CTAS with date partition strategy

For tables that grow daily, avoid full-table CTAS on every refresh — only regenerate the partitions that changed:

from datetime import date, timedelta

def refresh_missing_partitions(
    start_date: date,
    end_date: date,
) -> list[str]:
    """Materialize only the date partitions not yet in the summary table."""
    glue = boto3.client("glue", region_name="us-east-1")

    # List partitions already materialized
    paginator = glue.get_paginator("get_partitions")
    existing_partitions = set()
    for page in paginator.paginate(
        DatabaseName="analytics",
        TableName="mcp_daily_summary"
    ):
        for p in page["Partitions"]:
            existing_partitions.add(p["Values"][0])   # dt value

    # Find gaps
    current = start_date
    missing = []
    while current <= end_date:
        if current.isoformat() not in existing_partitions:
            missing.append(current)
        current += timedelta(days=1)

    # Materialize missing partitions
    for dt in missing:
        materialize_daily_summary(dt)

    return [d.isoformat() for d in missing]

# Called from an MCP admin tool or daily schedule Lambda
refreshed = refresh_missing_partitions(
    start_date=date(2026, 9, 1),
    end_date=date.today() - timedelta(days=1),   # yesterday's data is complete
)

Incremental partition refresh means CTAS costs scale with new data volume, not with total table history. A table with two years of daily partitions still costs the same to refresh as a table with one week of history — each refresh only touches the new partition's source data. This makes CTAS-based materialization sustainable as the MCP analytics corpus grows.