CREATE TABLE test_welford (
    `id` INT PRIMARY KEY,
    `value` DOUBLE,
    `ts` TIMESTAMP TIME INDEX DEFAULT now()
);

INSERT INTO test_welford (`id`, `value`) VALUES
    (1, 10.0),
    (2, 20.0),
    (3, 30.0),
    (4, 40.0),
    (5, 50.0),
    (6, 60.0),
    (7, 70.0),
    (8, 80.0),
    (9, 90.0),
    (10, 100.0),
    (11, NULL);

SELECT stddev_pop_calc(stddev_pop_state(`value`)) FROM test_welford;

-- A second DISTINCT argument set prevents DataFusion from rewriting this to GROUP BY.
SELECT
    stddev_pop_calc(stddev_pop_state(DISTINCT `value`)),
    count(DISTINCT `id`)
FROM (
    SELECT `id`, `value` FROM test_welford
    UNION ALL
    SELECT 12 AS `id`, 100.0 AS `value`
) AS duplicated_welford;

SELECT stddev_pop_calc(stddev_pop_state(`value`)) FROM test_welford WHERE false;

CREATE TABLE grouped_welford (
    `id` INT PRIMARY KEY,
    `state` BINARY,
    `ts` TIMESTAMP TIME INDEX DEFAULT now()
);

INSERT INTO grouped_welford (`id`, `state`)
SELECT 1, stddev_pop_state(`value`) FROM test_welford WHERE id <= 5;

INSERT INTO grouped_welford (`id`, `state`)
SELECT 2, stddev_pop_state(`value`) FROM test_welford WHERE id > 5;

SELECT stddev_pop_calc(stddev_pop_merge(`state`)) FROM grouped_welford;

-- A second DISTINCT argument set prevents DataFusion from rewriting this to GROUP BY.
SELECT
    stddev_pop_calc(stddev_pop_merge(DISTINCT `state`)),
    count(DISTINCT `id`)
FROM (
    SELECT `id`, `state` FROM grouped_welford
    UNION ALL
    SELECT 3 AS `id`, `state` FROM grouped_welford WHERE id = 1
) AS duplicated_states;

CREATE TABLE welford_delta_merge (
    `id` INT PRIMARY KEY,
    `delta` BINARY,
    `persisted` BINARY,
    `ts` TIMESTAMP TIME INDEX DEFAULT now()
);

INSERT INTO welford_delta_merge (`id`, `delta`, `persisted`)
SELECT 1, stddev_pop_state(`value`), NULL
FROM test_welford WHERE id <= 5;

INSERT INTO welford_delta_merge (`id`, `delta`, `persisted`)
SELECT 2, NULL, stddev_pop_state(`value`)
FROM test_welford WHERE id > 5;

INSERT INTO welford_delta_merge (`id`, `delta`, `persisted`)
VALUES (3, NULL, NULL);

-- Delta-only, persisted-only, and both-null (empty-state) identity behavior.
SELECT
    `id`,
    stddev_pop_calc(__stddev_pop_state_delta_merge(`delta`, `persisted`)) AS merged_stddev
FROM welford_delta_merge
GROUP BY `id`
ORDER BY `id`;

-- Merging delta and persisted states must equal direct stddev_pop state calculation.
SELECT
    stddev_pop_calc(__stddev_pop_state_delta_merge(`delta`, `persisted`)) AS merged_stddev,
    (SELECT stddev_pop_calc(stddev_pop_state(`value`)) FROM test_welford) AS direct_stddev
FROM welford_delta_merge;

DROP TABLE welford_delta_merge;
DROP TABLE grouped_welford;
DROP TABLE test_welford;

CREATE TABLE welford_window_raw (
    `id` INT PRIMARY KEY,
    `value` DOUBLE,
    `ts` TIMESTAMP TIME INDEX
);

INSERT INTO welford_window_raw VALUES
    (1, 1.0, '2024-01-01 00:01:05'),
    (2, 2.0, '2024-01-01 00:01:20'),
    (3, 3.0, '2024-01-01 00:01:50'),
    (4, 10.0, '2024-01-01 00:02:10'),
    (5, 20.0, '2024-01-01 00:02:40'),
    (6, 4.0, '2024-01-01 00:03:05'),
    (7, 8.0, '2024-01-01 00:03:15'),
    (8, 12.0, '2024-01-01 00:03:35'),
    (9, 16.0, '2024-01-01 00:03:55'),
    (10, 100.0, '2024-01-01 00:04:05'),
    (11, 200.0, '2024-01-01 00:04:25'),
    (12, 300.0, '2024-01-01 00:04:45');

CREATE TABLE welford_minute_states (
    `minute_ts` TIMESTAMP TIME INDEX,
    `state` BINARY
);

INSERT INTO welford_minute_states (`minute_ts`, `state`)
SELECT
    date_bin(INTERVAL '1 minute', `ts`) AS minute_ts,
    stddev_pop_state(`value`) AS state
FROM welford_window_raw
GROUP BY minute_ts;

-- Merging persisted minute states must reproduce aggregation over the raw samples.
WITH ranges AS (
    SELECT
        '1-3' AS range_name,
        CAST('2024-01-01 00:01:00' AS TIMESTAMP) AS start_ts,
        CAST('2024-01-01 00:04:00' AS TIMESTAMP) AS end_ts
    UNION ALL
    SELECT
        '2-4' AS range_name,
        CAST('2024-01-01 00:02:00' AS TIMESTAMP) AS start_ts,
        CAST('2024-01-01 00:05:00' AS TIMESTAMP) AS end_ts
), direct AS (
    SELECT
        ranges.range_name,
        count(*) AS sample_count,
        stddev_pop(raw.`value`) AS stddev
    FROM ranges CROSS JOIN welford_window_raw AS raw
    WHERE raw.`ts` >= ranges.start_ts
      AND raw.`ts` < ranges.end_ts
    GROUP BY ranges.range_name
), merged AS (
    SELECT
        ranges.range_name,
        count(*) AS state_count,
        stddev_pop_calc(stddev_pop_merge(states.`state`)) AS stddev
    FROM ranges CROSS JOIN welford_minute_states AS states
    WHERE states.minute_ts >= ranges.start_ts
      AND states.minute_ts < ranges.end_ts
    GROUP BY ranges.range_name
)
SELECT
    direct.range_name,
    direct.sample_count,
    merged.state_count,
    direct.stddev AS direct_stddev,
    merged.stddev AS merged_stddev,
    abs(direct.stddev - merged.stddev) AS difference
FROM direct JOIN merged ON direct.range_name = merged.range_name
ORDER BY direct.range_name;

DROP TABLE welford_minute_states;
DROP TABLE welford_window_raw;