-- ============================================================================
-- AWS Lambda CloudWatch Logs -> ClickHouse Cloud via S3Queue
-- ============================================================================
-- This schema is for Lambda logs exported from CloudWatch Logs to S3.
-- Lambda logs are plain text with embedded timestamps and optional JSON data.
--
-- Log format: TIMESTAMP MESSAGE
-- Example: 2026-01-19T20:50:10.108Z INIT_START Runtime Version: provided:al2023.v117
-- ============================================================================

-- ============================================================================
-- 1. DESTINATION TABLE (MergeTree)
-- ============================================================================
-- This table stores the parsed Lambda logs for querying.
-- We extract JSON from log messages and parse REPORT lines for metrics.

CREATE TABLE IF NOT EXISTS lambda_logs
(
    -- Timestamp from the log line (ISO 8601 format)
    timestamp DateTime64(3) DEFAULT now64(),

    -- Lambda function name (extracted from S3 path)
    function_name LowCardinality(String) DEFAULT '',

    -- CloudWatch log stream ID (execution environment identifier)
    -- Optional: helps identify container reuse and concurrent execution patterns
    stream_id String DEFAULT '',

    -- Raw log message (everything after the timestamp)
    message String,

    -- Log type classification
    log_type LowCardinality(String) DEFAULT '',

    -- Request ID (extracted from START/END/REPORT lines or custom logs)
    request_id String DEFAULT '',

    -- Extracted JSON data (for logs with embedded JSON)
    -- Using JSON type for dynamic schema and simpler querying
    -- Nullable because not all log lines contain JSON
    json_data Nullable(JSON),

    -- REPORT line metrics (only populated for REPORT log type)
    duration_ms Float64 DEFAULT 0,
    billed_duration_ms UInt32 DEFAULT 0,
    memory_size_mb UInt32 DEFAULT 0,
    max_memory_used_mb UInt32 DEFAULT 0,
    init_duration_ms Float64 DEFAULT 0,

    -- Custom application log fields (extracted from structured logs)
    http_method LowCardinality(String) DEFAULT '',
    http_path String DEFAULT '',
    http_duration_us Float64 DEFAULT 0,
    client_ip String DEFAULT '',

    -- Materialized columns for common query patterns
    date Date MATERIALIZED toDate(timestamp),
    hour UInt8 MATERIALIZED toHour(timestamp),
    is_cold_start UInt8 MATERIALIZED log_type = 'INIT_START',
    is_error UInt8 MATERIALIZED position(message, 'ERROR') > 0 OR position(message, 'error') > 0,
    has_json UInt8 MATERIALIZED isNotNull(json_data)
)
ENGINE = MergeTree
PARTITION BY toYYYYMMDD(timestamp)
ORDER BY (function_name, timestamp, request_id)
TTL timestamp + INTERVAL 90 DAY
SETTINGS index_granularity = 8192;


-- ============================================================================
-- 2. S3QUEUE TABLE (Streaming Source)
-- ============================================================================
-- This table continuously polls S3 for new Lambda log files exported from CloudWatch.
--
-- IMPORTANT: Replace the following placeholders:
--   - YOUR_BUCKET_NAME: Your S3 bucket name
--   - YOUR_REGION: AWS region (e.g., us-east-1)
--   - YOUR_IAM_ROLE_ARN: IAM role ARN for ClickHouse to assume
--
-- Path pattern for Lambda logs:
--   s3://bucket/lambda/**.gz  (recursive search in lambda/ prefix)
--
-- Multi-function support:
--   The function name is auto-extracted from the S3 path.
--   Common patterns:
--     lambda/function-name/2026/01/19/stream-id/file.gz → function_name = "function-name"
--     lambda/exports/aws-lambda-func/... → function_name = "exports"
--   All functions' logs go into the same table with function_name field for filtering.
--
-- Mode selection:
--   Using 'unordered' mode instead of 'ordered' because:
--   - CloudWatch can delay log delivery up to 24 hours
--   - Multiple functions create complex lexicographic ordering (lambda/func-a/..., lambda/func-b/...)
--   - Multiple streams within each function add more path variations
--   - Unordered mode tracks each file individually, handles late arrivals correctly
--   - Slightly more metadata overhead, but much more reliable for this use case

CREATE TABLE IF NOT EXISTS lambda_logs_queue
(
    -- Raw log line as a single string
    raw_line String
)
ENGINE = S3Queue(
    'https://<BUCKET_NAME>.s3.<REGION>.amazonaws.com/lambda/**.gz',
    'LineAsString',
    'gzip',
    extra_credentials(role_arn = 'arn:aws:iam::012345678901:role/ClickHouseAccess-foobar')
)
SETTINGS
    mode = 'unordered',
    s3queue_loading_retries = 3,
    -- Lambda logs export to S3 on a schedule (hourly exports)
    -- Poll every 5-10 minutes since we don't need real-time and want to minimize API costs
    s3queue_polling_min_timeout_ms = 300000,   -- 5 minutes
    s3queue_polling_max_timeout_ms = 600000,   -- 10 minutes
    s3queue_enable_logging_to_s3queue_log = 1;


-- ============================================================================
-- 3. MATERIALIZED VIEW (Connects Queue to Destination)
-- ============================================================================
-- This view transforms raw log lines into structured data.
--
-- Key transformations:
--   - Extracts timestamp (first 24 chars: "2026-01-19T20:50:10.108Z")
--   - Classifies log type (INIT_START, START, END, REPORT, or application log)
--   - Extracts RequestId from START/END/REPORT lines
--   - Parses REPORT metrics into separate columns
--   - Extracts embedded JSON from log messages
--   - Parses custom application logs for HTTP method, path, duration

CREATE MATERIALIZED VIEW IF NOT EXISTS lambda_logs_mv
TO lambda_logs
AS SELECT
    -- Parse timestamp from first 24 characters
    parseDateTime64BestEffortOrZero(substring(raw_line, 1, 24), 3, 'UTC') AS timestamp,

    -- Extract Lambda function name from S3 path
    -- Common patterns:
    --   lambda/function-name/2026/01/19/file.gz → function-name
    --   lambda/aws-logs-export/aws-lambda-function-name/2026/... → aws-lambda-function-name
    -- Takes the segment after 'lambda/' prefix (before next /)
    arrayElement(splitByChar('/', _path),
        arrayFirstIndex(x -> x = 'lambda', splitByChar('/', _path)) + 1) AS function_name,

    -- Extract CloudWatch log stream ID from S3 path
    -- CloudWatch export includes stream as parent directory of the file:
    --   lambda/function/2026/01/19/uuid/2026-01-19-[$LATEST]abc123/000000.gz → 2026-01-19-[$LATEST]abc123
    -- The stream ID typically contains [$LATEST] or similar Lambda version markers
    -- Get the second-to-last path segment (parent directory of file)
    arrayElement(splitByChar('/', _path), length(splitByChar('/', _path)) - 1) AS stream_id,

    -- Extract message (everything after timestamp + space)
    trimLeft(substring(raw_line, 26)) AS message,

    -- Classify log type
    multiIf(
        position(message, 'INIT_START') = 1, 'INIT_START',
        position(message, 'START RequestId:') = 1, 'START',
        position(message, 'END RequestId:') = 1, 'END',
        position(message, 'REPORT RequestId:') = 1, 'REPORT',
        position(message, 'ERROR') > 0, 'ERROR',
        position(message, 'WARN') > 0, 'WARN',
        'INFO'
    ) AS log_type,

    -- Extract RequestId from START/END/REPORT lines
    multiIf(
        position(message, 'START RequestId:') = 1,
            trim(substring(message, position(message, 'RequestId:') + 11, position(message, ' Version:') - position(message, 'RequestId:') - 11)),
        position(message, 'END RequestId:') = 1,
            trim(substring(message, position(message, 'RequestId:') + 11)),
        position(message, 'REPORT RequestId:') = 1,
            trim(substring(message, position(message, 'RequestId:') + 11, position(message, '\t') - position(message, 'RequestId:') - 11)),
        ''
    ) AS request_id,

    -- Extract JSON data (look for { ... } patterns and cast to JSON type)
    if(
        position(message, '{') > 0 AND position(message, '}') > 0,
        substring(message, position(message, '{'), position(message, '}') - position(message, '{') + 1)::JSON,
        NULL
    ) AS json_data,

    -- Parse REPORT metrics
    if(log_type = 'REPORT',
        toFloat64OrZero(
            trim(substring(message,
                position(message, 'Duration:') + 10,
                position(message, ' ms\t') - position(message, 'Duration:') - 10
            ))
        ),
        0
    ) AS duration_ms,

    if(log_type = 'REPORT',
        toUInt32OrZero(
            trim(substring(message,
                position(message, 'Billed Duration:') + 17,
                positionCaseInsensitive(message, ' ms\t', position(message, 'Billed Duration:')) - position(message, 'Billed Duration:') - 17
            ))
        ),
        0
    ) AS billed_duration_ms,

    if(log_type = 'REPORT',
        toUInt32OrZero(
            trim(substring(message,
                position(message, 'Memory Size:') + 13,
                positionCaseInsensitive(message, ' MB\t', position(message, 'Memory Size:')) - position(message, 'Memory Size:') - 13
            ))
        ),
        0
    ) AS memory_size_mb,

    if(log_type = 'REPORT',
        toUInt32OrZero(
            trim(substring(message,
                position(message, 'Max Memory Used:') + 17,
                positionCaseInsensitive(message, ' MB\t', position(message, 'Max Memory Used:')) - position(message, 'Max Memory Used:') - 17
            ))
        ),
        0
    ) AS max_memory_used_mb,

    if(log_type = 'REPORT' AND position(message, 'Init Duration:') > 0,
        toFloat64OrZero(
            trim(substring(message,
                position(message, 'Init Duration:') + 15,
                positionCaseInsensitive(message, ' ms', position(message, 'Init Duration:')) - position(message, 'Init Duration:') - 15
            ))
        ),
        0
    ) AS init_duration_ms,

    -- Parse custom application logs with HTTP method pattern
    if(positionCaseInsensitiveUTF8(message, ' GET /') > 0 OR
       positionCaseInsensitiveUTF8(message, ' POST /') > 0 OR
       positionCaseInsensitiveUTF8(message, ' PUT /') > 0 OR
       positionCaseInsensitiveUTF8(message, ' DELETE /') > 0 OR
       positionCaseInsensitiveUTF8(message, ' OPTIONS /') > 0 OR
       positionCaseInsensitiveUTF8(message, ' PATCH /') > 0,
        arrayElement(splitByWhitespace(message), arrayFirstIndex(x -> x IN ('GET', 'POST', 'PUT', 'DELETE', 'OPTIONS', 'PATCH'), splitByWhitespace(message))),
        ''
    ) AS http_method,

    if(http_method != '',
        arrayElement(splitByWhitespace(message), arrayFirstIndex(x -> x IN ('GET', 'POST', 'PUT', 'DELETE', 'OPTIONS', 'PATCH'), splitByWhitespace(message)) + 1),
        ''
    ) AS http_path,

    if(position(message, 'Took:') > 0,
        multiIf(
            position(message, 'µs') > 0,
                toFloat64OrZero(trim(substring(message, position(message, 'Took:') + 6, position(message, 'µs') - position(message, 'Took:') - 6))),
            position(message, 'ms') > 0,
                toFloat64OrZero(trim(substring(message, position(message, 'Took:') + 6, position(message, 'ms') - position(message, 'Took:') - 6))) * 1000,
            position(message, 's') > 0 AND position(message, 'µs') = 0 AND position(message, 'ms') = 0,
                toFloat64OrZero(trim(substring(message, position(message, 'Took:') + 6, position(message, 's') - position(message, 'Took:') - 6))) * 1000000,
            0
        ),
        0
    ) AS http_duration_us,

    -- Extract client IP (element before HTTP method in log format)
    -- Format: YYYY/MM/DD HH:MM:SS IP METHOD PATH Took: Xµs
    if(http_method != '',
        arrayElement(splitByWhitespace(message),
            arrayFirstIndex(x -> x IN ('GET', 'POST', 'PUT', 'DELETE', 'OPTIONS', 'PATCH'), splitByWhitespace(message)) - 1),
        ''
    ) AS client_ip

FROM lambda_logs_queue
WHERE length(raw_line) > 24;  -- Skip empty lines


-- ============================================================================
-- 4. USEFUL QUERIES
-- ============================================================================

-- List all Lambda functions in the data
-- SELECT
--     function_name,
--     count() AS log_lines,
--     countIf(log_type = 'START') AS invocations
-- FROM lambda_logs
-- WHERE date >= today() - 7
-- GROUP BY function_name
-- ORDER BY invocations DESC;

-- Request count by hour (for specific function)
-- SELECT
--     toStartOfHour(timestamp) AS hour,
--     count() AS total_requests,
--     sum(is_cold_start) AS cold_starts,
--     sum(is_error) AS errors
-- FROM lambda_logs
-- WHERE date = today()
--   AND function_name = 'roi-calculator'
--   AND log_type = 'START'
-- GROUP BY hour
-- ORDER BY hour;

-- Compare performance across all functions
-- SELECT
--     function_name,
--     count() AS invocations,
--     quantile(0.5)(duration_ms) AS p50_ms,
--     quantile(0.9)(duration_ms) AS p90_ms,
--     quantile(0.99)(duration_ms) AS p99_ms,
--     avg(max_memory_used_mb) AS avg_memory_mb
-- FROM lambda_logs
-- WHERE date = today() AND log_type = 'REPORT'
-- GROUP BY function_name
-- ORDER BY invocations DESC;

-- Lambda performance metrics (from REPORT lines)
-- SELECT
--     quantile(0.5)(duration_ms) AS p50_duration_ms,
--     quantile(0.9)(duration_ms) AS p90_duration_ms,
--     quantile(0.99)(duration_ms) AS p99_duration_ms,
--     avg(max_memory_used_mb) AS avg_memory_mb,
--     max(max_memory_used_mb) AS max_memory_mb
-- FROM lambda_logs
-- WHERE date = today() AND log_type = 'REPORT';

-- Cold start analysis
-- SELECT
--     toStartOfHour(timestamp) AS hour,
--     countIf(init_duration_ms > 0) AS cold_starts,
--     avg(init_duration_ms) AS avg_init_ms,
--     quantile(0.9)(init_duration_ms) AS p90_init_ms
-- FROM lambda_logs
-- WHERE date = today() AND log_type = 'REPORT'
-- GROUP BY hour
-- ORDER BY hour;

-- Container reuse analysis (using stream_id)
-- Each stream represents an execution environment (container)
-- SELECT
--     function_name,
--     uniq(stream_id) AS unique_containers,
--     count() AS total_invocations,
--     round(count() / uniq(stream_id), 2) AS avg_invocations_per_container
-- FROM lambda_logs
-- WHERE date = today() AND log_type = 'START' AND stream_id != ''
-- GROUP BY function_name
-- ORDER BY unique_containers DESC;

-- Identify concurrent executions (multiple streams active at same time)
-- SELECT
--     function_name,
--     toStartOfMinute(timestamp) AS minute,
--     uniq(stream_id) AS concurrent_containers
-- FROM lambda_logs
-- WHERE date = today() AND log_type = 'START' AND stream_id != ''
-- GROUP BY function_name, minute
-- HAVING concurrent_containers > 1
-- ORDER BY concurrent_containers DESC
-- LIMIT 20;

-- Trace all requests handled by a specific container
-- SELECT
--     timestamp,
--     request_id,
--     log_type,
--     duration_ms,
--     init_duration_ms
-- FROM lambda_logs
-- WHERE stream_id = '[$LATEST]abc123-def456'
--   AND log_type = 'REPORT'
-- ORDER BY timestamp;

-- HTTP endpoint analysis (from custom application logs)
-- SELECT
--     http_method,
--     http_path,
--     count() AS requests,
--     quantile(0.5)(http_duration_us / 1000) AS p50_ms,
--     quantile(0.9)(http_duration_us / 1000) AS p90_ms,
--     quantile(0.99)(http_duration_us / 1000) AS p99_ms
-- FROM lambda_logs
-- WHERE date = today() AND http_method != ''
-- GROUP BY http_method, http_path
-- ORDER BY requests DESC;

-- Extract and analyze JSON data (using JSON type for simpler syntax)
-- SELECT
--     timestamp,
--     request_id,
--     message,
--     json_data.event AS event_type,
--     json_data.observabilitySpend AS observability_spend,
--     json_data.savings3Year AS savings_3year,
--     json_data.ipAddress AS ip
-- FROM lambda_logs
-- WHERE date = today() AND has_json = 1
-- LIMIT 100;
--
-- Discover all JSON fields in the data:
-- DESCRIBE TABLE lambda_logs;

-- Trace individual requests across all log lines
-- SELECT
--     timestamp,
--     log_type,
--     message,
--     duration_ms,
--     max_memory_used_mb
-- FROM lambda_logs
-- WHERE request_id = 'f6499439-8583-43e7-bdcb-b43eba4ed50a'
-- ORDER BY timestamp;

-- Error rate by hour
-- SELECT
--     toStartOfHour(timestamp) AS hour,
--     countIf(is_error = 1) AS errors,
--     count() AS total_logs,
--     round(countIf(is_error = 1) / count() * 100, 2) AS error_rate_pct
-- FROM lambda_logs
-- WHERE date = today()
-- GROUP BY hour
-- ORDER BY hour;

-- Top client IPs by request count
-- SELECT
--     client_ip,
--     count() AS requests,
--     uniq(request_id) AS unique_requests
-- FROM lambda_logs
-- WHERE date = today() AND client_ip != ''
-- GROUP BY client_ip
-- ORDER BY requests DESC
-- LIMIT 20;
