%
WITH
    {{
        Int16(
            max_rate_of_change,
            description="Maximum allowable rate of change from point to point",
            required=True,
        )
    }} as _max_rate_of_change,
    calculate_rate_of_change AS (
        SELECT
            id,
            timestamp,
            previous_timestamp,
            (value - previous_value) / (timestamp - previous_timestamp) AS rate_of_change,
            value,
            previous_value,
            (value - previous_value) AS value_diff,
            (timestamp - previous_timestamp) AS time_diff,
            lagInFrame(timestamp, 1) OVER (
                PARTITION BY id ORDER BY timestamp ASC ROWS BETWEEN 1 PRECEDING AND 1 PRECEDING
            ) AS previous_timestamp,
            lagInFrame(value, 1) OVER (
                PARTITION BY id ORDER BY timestamp ASC ROWS BETWEEN 1 PRECEDING AND 1 PRECEDING
            ) AS previous_value
        FROM incoming_data
        ORDER BY timestamp DESC
    )
SELECT *
FROM calculate_rate_of_change
WHERE abs(rate_of_change) > _max_rate_of_change
ORDER BY timestamp DESC