Time series analysis and custom aggregate function for Postgres

Question

I have a table with time series. It contains snapshot and increment value

• snapshot means concrete value
• increment means take previous snapshot and add increment value to get snapshot value

sensor_id - id of sensor
ts - timestamp
temp - temperature at a given ts
value_type - 0 means snapshot, 1 means increment

sensor_id | ts | temp | value_type
sensor_X  |  1 |  100 | 0
sensor_X  |  2 |    5 | 1
sensor_X  |  3 |   -2 | 1
sensor_X  |  4 |   95 | 0

sensor_Y  |  4 |   90 | 0
sensor_Y  |  5 |   -5 | 1

My goal is to build time series view from table

           1|   2|   3|   4|    5| 
sensor_X  100| 105| 103|  95|    |     
sensor_Y     |    |    |  90|  85|  

user queries data by ts range and sensor_id:

show me sensor temperature where sensor_id starts with sensor_0* and ts in range (2023-01-20, 2023-01-23)

I try to implement aggregate function

-- thisValue, thisValueType, nextValue, nextValueType
CREATE OR REPLACE FUNCTION calculate_snapshot_internal(bigint, int, bigint, int)
    returns bigint
    language sql
as
$$
select case
           when $3 is null then null -- metric value explicitly set to null, means it removed
           when $4 = 1 or $4 = 2 then $3 -- next value is SNAPSHOT or FULL_SNAPSHOT, just set it
           when $4 = 0 and $1 is not null then $1 + $3 -- next value is INCREMENTAL, add it to current SNAPSHOT
           end
$$;

CREATE aggregate calculate_snapshot (bigint, int) (
    sfunc = calculate_snapshot_internal,
    stype = bigint
    -- initcond = null We can do this simply by omitting the initcond phrase, so that the initial state condition is null
    );

and query

select sensor_id,
       calculate_snapshot(value, valueType) OVER (PARTITION BY sensor_id ORDER BY ts) as snapshot_value
from metric_table;

I can't execute CREATE aggregate calculate_snapshot error:

[42883] ERROR: function calculate_snapshot_internal(bigint, bigint, integer) does not exist

Don't understand how is it supposed to work.

UPD: the error explanation is marked as correct answer. Here is final solution that worked for me

REATE OR REPLACE FUNCTION calculate_snapshot_internal(_this_value bigint,
                                                              _next_value bigint,
                                                              _next_value_type int,
                                                              OUT _code bigint)
    LANGUAGE plpgsql AS
$func$
BEGIN


    if _next_value is null then
        _code = _this_value;    -- just pass by whatever it has, null means missing value
    elseif _next_value_type = 1 or _next_value_type = 2 then -- SNAPSHOT value
        _code = _next_value;
    elseif _next_value_type = 0 and _this_value is not null then  -- Add INCREMENT value to accumulated value
        _code = _this_value + _next_value;
    end if;

END
$func$;

drop aggregate if exists calculate_snapshot(bigint, int);
CREATE aggregate calculate_snapshot(bigint, int) (
    sfunc = calculate_snapshot_internal,
    stype = bigint
    );

Answer 1

As the docs say, the signature of the sfunc must take the aggregate state type (stype) plus the aggregate function argument type or types. Since the stype is bigint and the aggregate types are (bigint, int), that makes the sfunc need (bigint, bigint, int), as the error message also indicates.

Since your sfunc never references $2, why does it declare to accept it? I can't understand what you want this to do (in detail), but the signature of the sfunc at least is clear given the other types in use.