-- Kafka-to-Spark Streaming Pipeline Validation Script
--
-- Purpose:
--   Use this SQL file as a safe, read-only reference for validating a Kafka-to-Spark
--   streaming pipeline. It focuses on source inspection, parsing expectations,
--   transformation checks, and output verification.
--
-- Notes:
--   - This script uses placeholders where engine-specific settings are required.
--   - Replace <...> values with your environment-specific names.
--   - Keep checkpointing and sink configuration in your Spark application code.
--   - Run against a test or staging environment first.

-- -----------------------------------------------------------------------------
-- 1) Define expected source schema for JSON events
-- -----------------------------------------------------------------------------
-- This is a logical schema reference. If your SQL engine supports inline schema
-- declarations, adapt this block to the syntax your runtime expects.

-- Expected event fields:
--   event_id   : unique stable identifier for deduplication and reconciliation
--   device_id  : source device or producer identifier
--   event_type : event category such as login, click, alert, or heartbeat
--   ts         : event timestamp in ISO-8601 or a consistently parseable format
--   status     : business status value such as ok, warn, or fail

-- -----------------------------------------------------------------------------
-- 2) Source inspection query
-- -----------------------------------------------------------------------------
-- Use this pattern to inspect raw Kafka records in a read-only way.
-- Adjust table/view names to match the integration layer in your environment.

-- Example logical source view:
--   <source_event_stream>
-- Columns assumed:
--   key, value, timestamp, partition, offset, topic

SELECT
    topic,
    partition,
    offset,
    timestamp,
    key,
    value
FROM <source_event_stream>
ORDER BY timestamp DESC
LIMIT 20;

-- Validation checklist:
-- - Confirm messages are arriving on the expected topic.
-- - Verify payloads are valid JSON strings.
-- - Ensure offsets advance when new events are produced.

-- -----------------------------------------------------------------------------
-- 3) Parsed record validation query
-- -----------------------------------------------------------------------------
-- This query shows the fields you should validate after parsing JSON payloads.
-- Replace the JSON extraction functions with those supported by your SQL engine.

WITH parsed_events AS (
    SELECT
        topic,
        partition,
        offset,
        timestamp AS kafka_timestamp,
        /* Replace these with your engine's JSON extraction functions */
        <json_extract_scalar(value, '$.event_id')>   AS event_id,
        <json_extract_scalar(value, '$.device_id')>  AS device_id,
        <json_extract_scalar(value, '$.event_type')> AS event_type,
        <json_extract_scalar(value, '$.ts')>         AS event_ts_raw,
        <json_extract_scalar(value, '$.status')>      AS status
    FROM <source_event_stream>
)
SELECT
    event_id,
    device_id,
    event_type,
    event_ts_raw,
    status,
    kafka_timestamp,
    partition,
    offset
FROM parsed_events
WHERE event_id IS NOT NULL
  AND device_id IS NOT NULL
ORDER BY kafka_timestamp DESC
LIMIT 50;

-- Validation checklist:
-- - Confirm required fields are present.
-- - Look for nulls in event_id, device_id, or ts.
-- - Check for inconsistent timestamp formats before production.

-- -----------------------------------------------------------------------------
-- 4) Data quality and malformed-record checks
-- -----------------------------------------------------------------------------
-- This section helps identify records that should be filtered or routed to a
-- controlled failure path in the streaming application.

WITH parsed_events AS (
    SELECT
        topic,
        partition,
        offset,
        timestamp AS kafka_timestamp,
        <json_extract_scalar(value, '$.event_id')>   AS event_id,
        <json_extract_scalar(value, '$.device_id')>  AS device_id,
        <json_extract_scalar(value, '$.event_type')> AS event_type,
        <json_extract_scalar(value, '$.ts')>         AS event_ts_raw,
        <json_extract_scalar(value, '$.status')>      AS status,
        value AS raw_value
    FROM <source_event_stream>
)
SELECT
    CASE
        WHEN raw_value IS NULL OR TRIM(raw_value) = '' THEN 'empty_payload'
        WHEN event_id IS NULL THEN 'missing_event_id'
        WHEN device_id IS NULL THEN 'missing_device_id'
        WHEN event_ts_raw IS NULL THEN 'missing_timestamp'
        WHEN status IS NULL THEN 'missing_status'
        ELSE 'ok'
    END AS record_quality,
    COUNT(*) AS record_count
FROM parsed_events
GROUP BY
    CASE
        WHEN raw_value IS NULL OR TRIM(raw_value) = '' THEN 'empty_payload'
        WHEN event_id IS NULL THEN 'missing_event_id'
        WHEN device_id IS NULL THEN 'missing_device_id'
        WHEN event_ts_raw IS NULL THEN 'missing_timestamp'
        WHEN status IS NULL THEN 'missing_status'
        ELSE 'ok'
    END
ORDER BY record_count DESC;

-- Validation checklist:
-- - Confirm malformed records are measurable.
-- - Confirm bad records are not silently treated as valid data.
-- - Make sure the application has a known handling strategy for failures.

-- -----------------------------------------------------------------------------
-- 5) Transformation readiness checks
-- -----------------------------------------------------------------------------
-- Use this query to verify downstream fields are suitable for analytics,
-- reconciliation, and alerting.

WITH normalized_events AS (
    SELECT
        <json_extract_scalar(value, '$.event_id')>   AS event_id,
        <json_extract_scalar(value, '$.device_id')>  AS device_id,
        <json_extract_scalar(value, '$.event_type')> AS event_type,
        <parse_timestamp_function(<json_extract_scalar(value, '$.ts')>)> AS event_ts,
        CASE
            WHEN LOWER(<json_extract_scalar(value, '$.status')>) = 'ok' THEN TRUE
            ELSE FALSE
        END AS is_success
    FROM <source_event_stream>
)
SELECT
    event_type,
    is_success,
    COUNT(*) AS total_events
FROM normalized_events
WHERE event_id IS NOT NULL
  AND device_id IS NOT NULL
  AND event_ts IS NOT NULL
GROUP BY event_type, is_success
ORDER BY total_events DESC;

-- Validation checklist:
-- - Confirm timestamps parse successfully.
-- - Confirm success flags match business rules.
-- - Verify grouping and filtering behave as expected.

-- -----------------------------------------------------------------------------
-- 6) Output reconciliation query
-- -----------------------------------------------------------------------------
-- Run this against the sink representation after the streaming job has written
-- data. Replace <output_event_sink> with the target table or externalized view.

SELECT
    event_id,
    device_id,
    event_type,
    event_ts,
    is_success
FROM <output_event_sink>
ORDER BY event_ts DESC
LIMIT 50;

-- Reconciliation example:
-- Compare source and sink counts over a recent window. Adjust windowing to match
-- your business SLA and event lateness tolerance.

WITH source_counts AS (
    SELECT COUNT(*) AS source_total
    FROM <source_event_stream>
    WHERE timestamp >= <window_start_timestamp>
),
output_counts AS (
    SELECT COUNT(*) AS sink_total
    FROM <output_event_sink>
    WHERE event_ts >= <window_start_timestamp>
)
SELECT
    source_total,
    sink_total,
    (source_total - sink_total) AS delta
FROM source_counts
CROSS JOIN output_counts;

-- Validation checklist:
-- - Confirm sink output is increasing as new events arrive.
-- - Compare source and sink totals for a recent window.
-- - Investigate any persistent delta before production promotion.

-- -----------------------------------------------------------------------------
-- 7) Operational preproduction checklist as SQL comments
-- -----------------------------------------------------------------------------
-- - Verify Kafka topic partition count supports your expected throughput.
-- - Confirm the Spark job uses a durable checkpoint location.
-- - Verify the sink tolerates duplicates or that deduplication is implemented.
-- - Confirm offset behavior is intentional (earliest for tests, controlled start
--   position for production).
-- - Confirm malformed events are filtered, quarantined, or clearly reported.
-- - Confirm restart behavior by stopping and starting the job in a test window.

-- End of validation script